Merge pull request ESQL-1260 from elastic/main

🤖 ESQL: Merge upstream

docs/changelog/96588.yaml

@@ -0,0 +1,5 @@
+pr: 96588
+summary: Support for patter_replace filter in keyword normalizer
+area: Search
+type: enhancement
+issues: []

docs/reference/analysis/normalizers.asciidoc

@@ -9,7 +9,7 @@ allowed, but not a stemming filter, which needs to look at the keyword as a
 whole. The current list of filters that can be used in a normalizer is
 following: `arabic_normalization`, `asciifolding`, `bengali_normalization`,
 `cjk_width`, `decimal_digit`, `elision`, `german_normalization`,
-`hindi_normalization`, `indic_normalization`, `lowercase`,
+`hindi_normalization`, `indic_normalization`, `lowercase`, `pattern_replace`,
 `persian_normalization`, `scandinavian_folding`, `serbian_normalization`,
 `sorani_normalization`, `uppercase`.
 

docs/reference/search-application/apis/delete-search-application.asciidoc

@@ -2,7 +2,7 @@
 [[delete-search-application]]
 === Delete Search Application
 
-preview::[]
+beta::[]
 
 ++++
 <titleabbrev>Delete Search Application</titleabbrev>

docs/reference/search-application/apis/get-search-application.asciidoc

@@ -2,7 +2,7 @@
 [[get-search-application]]
 === Get Search Application
 
-preview::[]
+beta::[]
 
 ++++
 <titleabbrev>Get Search Application</titleabbrev>

docs/reference/search-application/apis/index.asciidoc

@@ -1,7 +1,7 @@
 [[search-application-apis]]
 == Search Application APIs
 
-preview::[]
+beta::[]
 
 ++++
 <titleabbrev>Search Application APIs</titleabbrev>

docs/reference/search-application/apis/list-search-applications.asciidoc

@@ -2,7 +2,7 @@
 [[list-search-applications]]
 === List Search Applications
 
-preview::[]
+beta::[]
 
 ++++
 <titleabbrev>List Search Applications</titleabbrev>

docs/reference/search-application/apis/put-search-application.asciidoc

@@ -2,7 +2,7 @@
 [[put-search-application]]
 === Put Search Application
 
-preview::[]
+beta::[]
 
 ++++
 <titleabbrev>Put Search Application</titleabbrev>

docs/reference/search-application/apis/search-application-search.asciidoc

@@ -2,7 +2,7 @@
 [[search-application-search]]
 === Search Application Search
 
-preview::[]
+beta::[]
 
 ++++
 <titleabbrev>Search Application Search</titleabbrev>

modules/analysis-common/src/main/java/org/elasticsearch/analysis/common/PatternReplaceTokenFilterFactory.java

@@ -15,10 +15,11 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.analysis.AbstractTokenFilterFactory;
+import org.elasticsearch.index.analysis.NormalizingTokenFilterFactory;
 
 import java.util.regex.Pattern;
 
-public class PatternReplaceTokenFilterFactory extends AbstractTokenFilterFactory {
+public class PatternReplaceTokenFilterFactory extends AbstractTokenFilterFactory implements NormalizingTokenFilterFactory {
 
     private final Pattern pattern;
     private final String replacement;

modules/analysis-common/src/test/java/org/elasticsearch/analysis/common/PatternReplaceTokenFilterTests.java

@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+package org.elasticsearch.analysis.common;
+
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.index.analysis.AnalysisTestsHelper;
+import org.elasticsearch.index.analysis.NamedAnalyzer;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.ESTokenStreamTestCase;
+
+import java.io.IOException;
+
+public class PatternReplaceTokenFilterTests extends ESTokenStreamTestCase {
+
+    public void testNormalizer() throws IOException {
+        Settings settings = Settings.builder()
+            .putList("index.analysis.normalizer.my_normalizer.filter", "replace_zeros")
+            .put("index.analysis.filter.replace_zeros.type", "pattern_replace")
+            .put("index.analysis.filter.replace_zeros.pattern", "0+")
+            .put("index.analysis.filter.replace_zeros.replacement", "")
+            .put("index.analysis.filter.replace_zeros.all", true)
+            .put(Environment.PATH_HOME_SETTING.getKey(), createTempDir().toString())
+            .build();
+        ESTestCase.TestAnalysis analysis = AnalysisTestsHelper.createTestAnalysisFromSettings(settings, new CommonAnalysisPlugin());
+        assertNull(analysis.indexAnalyzers.get("my_normalizer"));
+        NamedAnalyzer normalizer = analysis.indexAnalyzers.getNormalizer("my_normalizer");
+        assertNotNull(normalizer);
+        assertEquals("my_normalizer", normalizer.name());
+        assertTokenStreamContents(normalizer.tokenStream("foo", "0000111"), new String[] { "111" });
+        assertEquals(new BytesRef("111"), normalizer.normalize("foo", "0000111"));
+    }
+
+}

modules/analysis-common/src/yamlRestTest/resources/rest-api-spec/test/analysis-common/40_token_filters.yml

@@ -1683,3 +1683,44 @@
     - length: { tokens: 6 }
     - match: { tokens.0.token: the }
     - match: { tokens.1.token: THE }
+
+---
+"pattern_replace_filter":
+  - do:
+      indices.create:
+        index: test
+        body:
+          settings:
+            analysis:
+              normalizer:
+                my_normalizer:
+                  type: custom
+                  filter: ["replace_zeros"]
+              filter:
+                replace_zeros:
+                  type: pattern_replace
+                  pattern: "0+"
+                  replacement: ""
+                  all: true
+          mappings:
+            properties:
+              pagerank:
+                type: keyword
+                normalizer: my_normalizer
+
+  - do:
+      index:
+        index:  test
+        id:     "1"
+        body:   { pagerank: "000000111"}
+
+  - do:
+      indices.refresh:
+        index: [ test ]
+
+  - do:
+      search:
+        index: test
+        q: pagerank:111
+
+  - match: {hits.total.value: 1}

server/src/internalClusterTest/java/org/elasticsearch/get/ShardMultiGetFomTranslogActionIT.java

@@ -34,6 +34,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 
 public class ShardMultiGetFomTranslogActionIT extends ESIntegTestCase {
+    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/96749")
     public void testShardMultiGetFromTranslog() throws Exception {
         assertAcked(
             prepareCreate("test").setSettings(

server/src/internalClusterTest/java/org/elasticsearch/index/shard/SearchIdleIT.java

@@ -16,6 +16,7 @@ import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.MultiGetRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
@@ -94,7 +95,9 @@ public class SearchIdleIT extends ESSingleNodeTestCase {
         client().prepareIndex("test").setId("0").setSource("{\"foo\" : \"bar\"}", XContentType.JSON).get();
         indexingDone.countDown(); // one doc is indexed above blocking
         IndexShard shard = indexService.getShard(0);
-        boolean hasRefreshed = shard.scheduledRefresh();
+        PlainActionFuture<Boolean> future = PlainActionFuture.newFuture();
+        shard.scheduledRefresh(future);
+        boolean hasRefreshed = future.actionGet();
         if (randomTimeValue == TimeValue.ZERO) {
             // with ZERO we are guaranteed to see the doc since we will wait for a refresh in the background
             assertFalse(hasRefreshed);
@@ -156,14 +159,14 @@ public class SearchIdleIT extends ESSingleNodeTestCase {
         ensureGreen();
         client().prepareIndex("test").setId("0").setSource("{\"foo\" : \"bar\"}", XContentType.JSON).get();
         IndexShard shard = indexService.getShard(0);
-        assertFalse(shard.scheduledRefresh());
+        scheduleRefresh(shard, false);
         assertTrue(shard.isSearchIdle());
         CountDownLatch refreshLatch = new CountDownLatch(1);
         // async on purpose to make sure it happens concurrently
         client().admin().indices().prepareRefresh().execute(ActionListener.running(refreshLatch::countDown));
         assertHitCount(client().prepareSearch().get(), 1);
         client().prepareIndex("test").setId("1").setSource("{\"foo\" : \"bar\"}", XContentType.JSON).get();
-        assertFalse(shard.scheduledRefresh());
+        scheduleRefresh(shard, false);
         assertTrue(shard.hasRefreshPending());
 
         // now disable background refresh and make sure the refresh happens
@@ -182,7 +185,7 @@ public class SearchIdleIT extends ESSingleNodeTestCase {
         // otherwise, it will compete to call `Engine#maybeRefresh` with the `scheduledRefresh` that we are going to verify.
         ensureNoPendingScheduledRefresh(indexService.getThreadPool());
         client().prepareIndex("test").setId("2").setSource("{\"foo\" : \"bar\"}", XContentType.JSON).get();
-        assertTrue(shard.scheduledRefresh());
+        scheduleRefresh(shard, true);
         assertFalse(shard.hasRefreshPending());
         assertTrue(shard.isSearchIdle());
         assertHitCount(client().prepareSearch().get(), 3);
@@ -194,6 +197,12 @@ public class SearchIdleIT extends ESSingleNodeTestCase {
         }
     }
 
+    private static void scheduleRefresh(IndexShard shard, boolean expectRefresh) {
+        PlainActionFuture<Boolean> future = PlainActionFuture.newFuture();
+        shard.scheduledRefresh(future);
+        assertThat(future.actionGet(), equalTo(expectRefresh));
+    }
+
     private void ensureNoPendingScheduledRefresh(ThreadPool threadPool) {
         // We can make sure that all scheduled refresh tasks are done by submitting *maximumPoolSize* blocking tasks,
         // then wait until all of them completed. Note that using ThreadPoolStats is not watertight as both queue and

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -15,6 +15,7 @@ import org.apache.lucene.search.Sort;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.util.Accountable;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
@@ -974,11 +975,17 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private void maybeRefreshEngine(boolean force) {
         if (indexSettings.getRefreshInterval().millis() > 0 || force) {
             for (IndexShard shard : this.shards.values()) {
-                try {
-                    shard.scheduledRefresh();
-                } catch (IndexShardClosedException | AlreadyClosedException ex) {
-                    // fine - continue;
-                }
+                shard.scheduledRefresh(new ActionListener<>() {
+                    @Override
+                    public void onResponse(Boolean ignored) {}
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        if (e instanceof IndexShardClosedException == false && e instanceof AlreadyClosedException == false) {
+                            logger.warn("unexpected exception while performing scheduled refresh", e);
+                        }
+                    }
+                });
             }
         }
     }

server/src/main/java/org/elasticsearch/index/analysis/PreConfiguredTokenFilter.java

@@ -120,13 +120,6 @@ public final class PreConfiguredTokenFilter extends PreConfiguredAnalysisCompone
         this.create = create;
     }
 
-    /**
-     * Can this {@link TokenFilter} be used in multi-term queries?
-     */
-    public boolean shouldUseFilterForMultitermQueries() {
-        return useFilterForMultitermQueries;
-    }
-
     @Override
     protected TokenFilterFactory create(Version version) {
         if (useFilterForMultitermQueries) {

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -1046,11 +1046,11 @@ public abstract class Engine implements Closeable {
     }
 
     /**
-     * Synchronously refreshes the engine for new search operations to reflect the latest
+     * Asynchronously refreshes the engine for new search operations to reflect the latest
      * changes unless another thread is already refreshing the engine concurrently.
      */
     @Nullable
-    public abstract RefreshResult maybeRefresh(String source) throws EngineException;
+    public abstract void maybeRefresh(String source, ActionListener<RefreshResult> listener) throws EngineException;
 
     /**
      * Called when our engine is using too much heap and should move buffered indexed/deleted documents to disk.

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -1944,15 +1944,15 @@ public class InternalEngine extends Engine {
     }
 
     @Override
-    public RefreshResult maybeRefresh(String source) throws EngineException {
-        return refresh(source, SearcherScope.EXTERNAL, false);
+    public void maybeRefresh(String source, ActionListener<RefreshResult> listener) throws EngineException {
+        ActionListener.completeWith(listener, () -> refresh(source, SearcherScope.EXTERNAL, false));
     }
 
     protected RefreshResult refreshInternalSearcher(String source, boolean block) throws EngineException {
         return refresh(source, SearcherScope.INTERNAL, block);
     }
 
-    final RefreshResult refresh(String source, SearcherScope scope, boolean block) throws EngineException {
+    protected final RefreshResult refresh(String source, SearcherScope scope, boolean block) throws EngineException {
         // both refresh types will result in an internal refresh but only the external will also
         // pass the new reader reference to the external reader manager.
         final long localCheckpointBeforeRefresh = localCheckpointTracker.getProcessedCheckpoint();
@@ -2113,7 +2113,7 @@ public class InternalEngine extends Engine {
                 // Only flush if (1) Lucene has uncommitted docs, or (2) forced by caller, or (3) the
                 // newly created commit points to a different translog generation (can free translog),
                 // or (4) the local checkpoint information in the last commit is stale, which slows down future recoveries.
-                boolean hasUncommittedChanges = indexWriter.hasUncommittedChanges();
+                boolean hasUncommittedChanges = hasUncommittedChanges();
                 if (hasUncommittedChanges
                     || force
                     || shouldPeriodicallyFlush()
@@ -2173,6 +2173,10 @@ public class InternalEngine extends Engine {
         waitForCommitDurability(generation, listener.map(v -> new FlushResult(true, generation)));
     }
 
+    protected boolean hasUncommittedChanges() {
+        return indexWriter.hasUncommittedChanges();
+    }
+
     private void refreshLastCommittedSegmentInfos() {
         /*
          * we have to inc-ref the store here since if the engine is closed by a tragic event

server/src/main/java/org/elasticsearch/index/engine/ReadOnlyEngine.java

@@ -434,8 +434,8 @@ public class ReadOnlyEngine extends Engine {
     }
 
     @Override
-    public RefreshResult maybeRefresh(String source) throws EngineException {
-        return RefreshResult.NO_REFRESH;
+    public void maybeRefresh(String source, ActionListener<RefreshResult> listener) throws EngineException {
+        listener.onResponse(RefreshResult.NO_REFRESH);
     }
 
     @Override

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -3738,11 +3738,9 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     }
 
     /**
-     * Executes a scheduled refresh if necessary.
-     *
-     * @return <code>true</code> iff the engine got refreshed otherwise <code>false</code>
+     * Executes a scheduled refresh if necessary. Completes the listener with true if a refreshed was performed otherwise false.
      */
-    public boolean scheduledRefresh() {
+    public void scheduledRefresh(ActionListener<Boolean> listener) {
         verifyNotClosed();
         boolean listenerNeedsRefresh = refreshListeners.refreshNeeded();
         if (isReadAllowed() && (listenerNeedsRefresh || getEngine().refreshNeeded())) {
@@ -3756,15 +3754,15 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
                 final Engine engine = getEngine();
                 engine.maybePruneDeletes(); // try to prune the deletes in the engine if we accumulated some
                 setRefreshPending(engine);
-                return false;
+                ActionListener.completeWith(listener, () -> false);
             } else {
                 logger.trace("refresh with source [schedule]");
-                return getEngine().maybeRefresh("schedule").refreshed();
+                getEngine().maybeRefresh("schedule", listener.map(Engine.RefreshResult::refreshed));
             }
         }
         final Engine engine = getEngine();
         engine.maybePruneDeletes(); // try to prune the deletes in the engine if we accumulated some
-        return false;
+        ActionListener.completeWith(listener, () -> false);
     }
 
     /**

server/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -47,6 +47,7 @@ import static org.elasticsearch.core.TimeValue.parseTimeValue;
 
 public class RestRequest implements ToXContent.Params {
 
+    public static final String RESPONSE_RESTRICTED = "responseRestricted";
     // tchar pattern as defined by RFC7230 section 3.2.6
     private static final Pattern TCHAR_PATTERN = Pattern.compile("[a-zA-Z0-9!#$%&'*+\\-.\\^_`|~]+");
 
@@ -612,6 +613,15 @@ public class RestRequest implements ToXContent.Params {
         return restApiVersion;
     }
 
+    public void markResponseRestricted(String restriction) {
+        if (params.containsKey(RESPONSE_RESTRICTED)) {
+            throw new IllegalArgumentException("The parameter [" + RESPONSE_RESTRICTED + "] is already defined.");
+        }
+        params.put(RESPONSE_RESTRICTED, restriction);
+        // this parameter is intended be consumed via ToXContent.Params.param(..), not this.params(..) so don't require it is consumed here
+        consumedParams.add(RESPONSE_RESTRICTED);
+    }
+
     public static class MediaTypeHeaderException extends RuntimeException {
 
         private final String message;

server/src/main/java/org/elasticsearch/rest/RestUtils.java

@@ -19,6 +19,8 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
 
+import static org.elasticsearch.rest.RestRequest.RESPONSE_RESTRICTED;
+
 public class RestUtils {
 
     /**
@@ -79,6 +81,9 @@ public class RestUtils {
     }
 
     private static void addParam(Map<String, String> params, String name, String value) {
+        if (RESPONSE_RESTRICTED.equalsIgnoreCase(name)) {
+            throw new IllegalArgumentException("parameter [" + RESPONSE_RESTRICTED + "] is reserved and may not set");
+        }
         params.put(name, value);
     }
 

server/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java

@@ -84,6 +84,7 @@ public class RestIndexAction extends BaseRestHandler {
         }
     }
 
+    @ServerlessScope(Scope.PUBLIC)
     public static final class AutoIdHandler extends RestIndexAction {
 
         private final Supplier<DiscoveryNodes> nodesInCluster;

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -3740,7 +3740,9 @@ public class IndexShardTests extends IndexShardTestCase {
         recoverShardFromStore(primary);
         indexDoc(primary, "_doc", "0", "{\"foo\" : \"bar\"}");
         assertTrue(primary.getEngine().refreshNeeded());
-        assertTrue(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future);
+        assertTrue(future.actionGet());
         assertFalse(primary.isSearchIdle());
 
         IndexScopedSettings scopedSettings = primary.indexSettings().getScopedSettings();
@@ -3787,7 +3789,9 @@ public class IndexShardTests extends IndexShardTestCase {
         recoverShardFromStore(primary);
         indexDoc(primary, "_doc", "0", "{\"foo\" : \"bar\"}");
         assertTrue(primary.getEngine().refreshNeeded());
-        assertTrue(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future);
+        assertTrue(future.actionGet());
         IndexScopedSettings scopedSettings = primary.indexSettings().getScopedSettings();
         settings = Settings.builder().put(settings).put(IndexSettings.INDEX_SEARCH_IDLE_AFTER.getKey(), TimeValue.ZERO).build();
         scopedSettings.applySettings(settings);
@@ -3796,7 +3800,9 @@ public class IndexShardTests extends IndexShardTestCase {
         indexDoc(primary, "_doc", "1", "{\"foo\" : \"bar\"}");
         assertTrue(primary.getEngine().refreshNeeded());
         long lastSearchAccess = primary.getLastSearcherAccess();
-        assertFalse(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future2 = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future2);
+        assertFalse(future2.actionGet());
         assertEquals(lastSearchAccess, primary.getLastSearcherAccess());
         // wait until the thread-pool has moved the timestamp otherwise we can't assert on this below
         assertBusy(() -> assertThat(primary.getThreadPool().relativeTimeInMillis(), greaterThan(lastSearchAccess)));
@@ -3821,7 +3827,9 @@ public class IndexShardTests extends IndexShardTestCase {
             assertEquals(1, searcher.getIndexReader().numDocs());
         }
         assertTrue(primary.getEngine().refreshNeeded());
-        assertTrue(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future3 = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future3);
+        assertTrue(future3.actionGet());
         latch.await();
         CountDownLatch latch1 = new CountDownLatch(1);
         primary.awaitShardSearchActive(refreshed -> {
@@ -3836,11 +3844,15 @@ public class IndexShardTests extends IndexShardTestCase {
         latch1.await();
 
         indexDoc(primary, "_doc", "2", "{\"foo\" : \"bar\"}");
-        assertFalse(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future4 = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future4);
+        assertFalse(future4.actionGet());
         assertTrue(primary.isSearchIdle());
         assertTrue(primary.searchIdleTime() >= TimeValue.ZERO.millis());
         primary.flushOnIdle(0);
-        assertTrue(primary.scheduledRefresh()); // make sure we refresh once the shard is inactive
+        PlainActionFuture<Boolean> future5 = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future5);
+        assertTrue(future5.actionGet()); // make sure we refresh once the shard is inactive
         try (Engine.Searcher searcher = primary.acquireSearcher("test")) {
             assertEquals(3, searcher.getIndexReader().numDocs());
         }
@@ -3855,7 +3867,9 @@ public class IndexShardTests extends IndexShardTestCase {
         recoverShardFromStore(primary);
         indexDoc(primary, "_doc", "0", "{\"foo\" : \"bar\"}");
         assertTrue(primary.getEngine().refreshNeeded());
-        assertTrue(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future);
+        assertTrue(future.actionGet());
         Engine.IndexResult doc = indexDoc(primary, "_doc", "1", "{\"foo\" : \"bar\"}");
         CountDownLatch latch = new CountDownLatch(1);
         if (randomBoolean()) {
@@ -3875,7 +3889,9 @@ public class IndexShardTests extends IndexShardTestCase {
         }
         assertEquals(1, latch.getCount());
         assertTrue(primary.getEngine().refreshNeeded());
-        assertTrue(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future2 = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future2);
+        assertTrue(future2.actionGet());
         latch.await();
 
         IndexScopedSettings scopedSettings = primary.indexSettings().getScopedSettings();
@@ -3891,7 +3907,9 @@ public class IndexShardTests extends IndexShardTestCase {
         }
         assertEquals(1, latch1.getCount());
         assertTrue(primary.getEngine().refreshNeeded());
-        assertTrue(primary.scheduledRefresh());
+        PlainActionFuture<Boolean> future3 = PlainActionFuture.newFuture();
+        primary.scheduledRefresh(future3);
+        assertTrue(future3.actionGet());
         latch1.await();
         closeShards(primary);
     }

server/src/test/java/org/elasticsearch/rest/RestRequestTests.java

@@ -31,8 +31,10 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
+import static org.elasticsearch.rest.RestRequest.RESPONSE_RESTRICTED;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -247,6 +249,22 @@ public class RestRequestTests extends ESTestCase {
         assertEquals("unknown content type", e.getMessage());
     }
 
+    public void testMarkResponseRestricted() {
+        RestRequest request1 = contentRestRequest("content", new HashMap<>());
+        request1.markResponseRestricted("foo");
+        assertEquals(request1.param(RESPONSE_RESTRICTED), "foo");
+        IllegalArgumentException exception = expectThrows(IllegalArgumentException.class, () -> request1.markResponseRestricted("foo"));
+        assertThat(exception.getMessage(), is("The parameter [" + RESPONSE_RESTRICTED + "] is already defined."));
+
+        RestRequest request2 = contentRestRequest("content", new HashMap<>() {
+            {
+                put(RESPONSE_RESTRICTED, "foo");
+            }
+        });
+        exception = expectThrows(IllegalArgumentException.class, () -> request2.markResponseRestricted("bar"));
+        assertThat(exception.getMessage(), is("The parameter [" + RESPONSE_RESTRICTED + "] is already defined."));
+    }
+
     public static RestRequest contentRestRequest(String content, Map<String, String> params) {
         Map<String, List<String>> headers = new HashMap<>();
         headers.put("Content-Type", Collections.singletonList("application/json"));

server/src/test/java/org/elasticsearch/rest/RestUtilsTests.java

@@ -16,6 +16,7 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.regex.Pattern;
 
+import static org.elasticsearch.rest.RestRequest.RESPONSE_RESTRICTED;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -156,6 +157,16 @@ public class RestUtilsTests extends ESTestCase {
         assertThat(params.size(), equalTo(1));
     }
 
+    public void testReservedParameters() {
+        Map<String, String> params = new HashMap<>();
+        String uri = "something?" + RESPONSE_RESTRICTED + "=value";
+        IllegalArgumentException exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> RestUtils.decodeQueryString(uri, uri.indexOf('?') + 1, params)
+        );
+        assertEquals(exception.getMessage(), "parameter [" + RESPONSE_RESTRICTED + "] is reserved and may not set");
+    }
+
     private void assertCorsSettingRegexIsNull(String settingsValue) {
         assertThat(RestUtils.checkCorsSettingForRegex(settingsValue), is(nullValue()));
     }

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/OperatorPrivilegesIT.java

@@ -97,18 +97,18 @@ public class OperatorPrivilegesIT extends ESRestTestCase {
     @SuppressWarnings("unchecked")
     public void testEveryActionIsEitherOperatorOnlyOrNonOperator() throws IOException {
         final String message = "An action should be declared to be either operator-only in ["
-            + OperatorOnlyRegistry.class.getName()
+            + DefaultOperatorOnlyRegistry.class.getName()
             + "] or non-operator in ["
             + Constants.class.getName()
             + "]";
 
-        Set<String> doubleLabelled = Sets.intersection(Constants.NON_OPERATOR_ACTIONS, OperatorOnlyRegistry.SIMPLE_ACTIONS);
+        Set<String> doubleLabelled = Sets.intersection(Constants.NON_OPERATOR_ACTIONS, DefaultOperatorOnlyRegistry.SIMPLE_ACTIONS);
         assertTrue("Actions are both operator-only and non-operator: [" + doubleLabelled + "]. " + message, doubleLabelled.isEmpty());
 
         final Request request = new Request("GET", "/_test/get_actions");
         final Map<String, Object> response = responseAsMap(client().performRequest(request));
         Set<String> allActions = Set.copyOf((List<String>) response.get("actions"));
-        final HashSet<String> labelledActions = new HashSet<>(OperatorOnlyRegistry.SIMPLE_ACTIONS);
+        final HashSet<String> labelledActions = new HashSet<>(DefaultOperatorOnlyRegistry.SIMPLE_ACTIONS);
         labelledActions.addAll(Constants.NON_OPERATOR_ACTIONS);
 
         final Set<String> unlabelled = Sets.difference(allActions, labelledActions);
@@ -119,7 +119,7 @@ public class OperatorPrivilegesIT extends ESRestTestCase {
             "Actions may no longer be valid: ["
                 + redundant
                 + "]. They should be removed from either the operator-only action registry in ["
-                + OperatorOnlyRegistry.class.getName()
+                + DefaultOperatorOnlyRegistry.class.getName()
                 + "] or the non-operator action list in ["
                 + Constants.class.getName()
                 + "]",

x-pack/plugin/security/src/main/java/module-info.java

@@ -63,6 +63,7 @@ module org.elasticsearch.security {
     exports org.elasticsearch.xpack.security.action.service to org.elasticsearch.server;
     exports org.elasticsearch.xpack.security.action.token to org.elasticsearch.server;
     exports org.elasticsearch.xpack.security.action.user to org.elasticsearch.server;
+    exports org.elasticsearch.xpack.security.operator to org.elasticsearch.internal.operator;
 
     exports org.elasticsearch.xpack.security.authc to org.elasticsearch.xcontent;
 

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java

@@ -295,10 +295,10 @@ import org.elasticsearch.xpack.security.authz.store.NativePrivilegeStore;
 import org.elasticsearch.xpack.security.authz.store.NativeRolesStore;
 import org.elasticsearch.xpack.security.authz.store.RoleProviders;
 import org.elasticsearch.xpack.security.ingest.SetSecurityUserProcessor;
+import org.elasticsearch.xpack.security.operator.DefaultOperatorOnlyRegistry;
 import org.elasticsearch.xpack.security.operator.FileOperatorUsersStore;
 import org.elasticsearch.xpack.security.operator.OperatorOnlyRegistry;
 import org.elasticsearch.xpack.security.operator.OperatorPrivileges;
-import org.elasticsearch.xpack.security.operator.OperatorPrivileges.OperatorPrivilegesService;
 import org.elasticsearch.xpack.security.profile.ProfileService;
 import org.elasticsearch.xpack.security.rest.RemoteHostHeader;
 import org.elasticsearch.xpack.security.rest.SecurityRestFilter;
@@ -543,6 +543,8 @@ public class Security extends Plugin
     private final List<SecurityExtension> securityExtensions = new ArrayList<>();
     private final SetOnce<Transport> transportReference = new SetOnce<>();
     private final SetOnce<ScriptService> scriptServiceReference = new SetOnce<>();
+    private final SetOnce<OperatorOnlyRegistry> operatorOnlyRegistry = new SetOnce<>();
+    private final SetOnce<OperatorPrivileges.OperatorPrivilegesService> operatorPrivilegesService = new SetOnce<>();
 
     private final SetOnce<ReservedRoleMappingAction> reservedRoleMappingAction = new SetOnce<>();
 
@@ -889,18 +891,26 @@ public class Security extends Plugin
         getLicenseState().addListener(allRolesStore::invalidateAll);
 
         final AuthenticationFailureHandler failureHandler = createAuthenticationFailureHandler(realms, extensionComponents);
-        final OperatorPrivilegesService operatorPrivilegesService;
-        final boolean operatorPrivilegesEnabled = OPERATOR_PRIVILEGES_ENABLED.get(settings);
+
+        // operator privileges are enabled either explicitly via the setting or if running serverless
+        final boolean operatorPrivilegesEnabled = OPERATOR_PRIVILEGES_ENABLED.get(settings) || DiscoveryNode.isServerless();
+
         if (operatorPrivilegesEnabled) {
             logger.info("operator privileges are enabled");
-            operatorPrivilegesService = new OperatorPrivileges.DefaultOperatorPrivilegesService(
-                getLicenseState(),
-                new FileOperatorUsersStore(environment, resourceWatcherService),
-                new OperatorOnlyRegistry(clusterService.getClusterSettings())
+            if (operatorOnlyRegistry.get() == null) {
+                operatorOnlyRegistry.set(new DefaultOperatorOnlyRegistry(clusterService.getClusterSettings()));
+            }
+            operatorPrivilegesService.set(
+                new OperatorPrivileges.DefaultOperatorPrivilegesService(
+                    getLicenseState(),
+                    new FileOperatorUsersStore(environment, resourceWatcherService),
+                    operatorOnlyRegistry.get()
+                )
             );
         } else {
-            operatorPrivilegesService = OperatorPrivileges.NOOP_OPERATOR_PRIVILEGES_SERVICE;
+            operatorPrivilegesService.set(OperatorPrivileges.NOOP_OPERATOR_PRIVILEGES_SERVICE);
         }
+
         authcService.set(
             new AuthenticationService(
                 settings,
@@ -912,7 +922,7 @@ public class Security extends Plugin
                 tokenService,
                 apiKeyService,
                 serviceAccountService,
-                operatorPrivilegesService
+                operatorPrivilegesService.get()
             )
         );
         components.add(authcService.get());
@@ -949,7 +959,7 @@ public class Security extends Plugin
             requestInterceptors,
             getLicenseState(),
             expressionResolver,
-            operatorPrivilegesService,
+            operatorPrivilegesService.get(),
             restrictedIndices
         );
 
@@ -1799,7 +1809,8 @@ public class Security extends Plugin
             secondayAuthc.get(),
             auditTrailService.get(),
             workflowService.get(),
-            handler
+            handler,
+            operatorPrivilegesService.get()
         );
     }
 
@@ -1903,6 +1914,19 @@ public class Security extends Plugin
     @Override
     public void loadExtensions(ExtensionLoader loader) {
         securityExtensions.addAll(loader.loadExtensions(SecurityExtension.class));
+
+        // operator registry SPI
+        List<OperatorOnlyRegistry> operatorOnlyRegistries = loader.loadExtensions(OperatorOnlyRegistry.class);
+        if (operatorOnlyRegistries.size() > 1) {
+            throw new IllegalStateException(OperatorOnlyRegistry.class + " may not have multiple implementations");
+        } else if (operatorOnlyRegistries.size() == 1) {
+            OperatorOnlyRegistry operatorOnlyRegistry = operatorOnlyRegistries.get(0);
+            this.operatorOnlyRegistry.set(operatorOnlyRegistry);
+            logger.debug(
+                "Loaded implementation [{}] for interface OperatorOnlyRegistry",
+                operatorOnlyRegistry.getClass().getCanonicalName()
+            );
+        }
     }
 
     private synchronized SharedGroupFactory getNettySharedGroupFactory(Settings settings) {
@@ -1945,4 +1969,9 @@ public class Security extends Plugin
         }
         return List.of(reservedRoleMappingAction.get());
     }
+
+    // visible for testing
+    OperatorPrivileges.OperatorPrivilegesService getOperatorPrivilegesService() {
+        return operatorPrivilegesService.get();
+    }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/operator/DefaultOperatorOnlyRegistry.java

@@ -0,0 +1,108 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.operator;
+
+import org.elasticsearch.action.admin.cluster.allocation.DeleteDesiredBalanceAction;
+import org.elasticsearch.action.admin.cluster.allocation.GetDesiredBalanceAction;
+import org.elasticsearch.action.admin.cluster.configuration.AddVotingConfigExclusionsAction;
+import org.elasticsearch.action.admin.cluster.configuration.ClearVotingConfigExclusionsAction;
+import org.elasticsearch.action.admin.cluster.desirednodes.DeleteDesiredNodesAction;
+import org.elasticsearch.action.admin.cluster.desirednodes.GetDesiredNodesAction;
+import org.elasticsearch.action.admin.cluster.desirednodes.UpdateDesiredNodesAction;
+import org.elasticsearch.action.admin.cluster.node.shutdown.PrevalidateNodeRemovalAction;
+import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsAction;
+import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.license.DeleteLicenseAction;
+import org.elasticsearch.license.PutLicenseAction;
+import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.transport.TransportRequest;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Stream;
+
+public class DefaultOperatorOnlyRegistry implements OperatorOnlyRegistry {
+
+    public static final Set<String> SIMPLE_ACTIONS = Set.of(
+        AddVotingConfigExclusionsAction.NAME,
+        ClearVotingConfigExclusionsAction.NAME,
+        PutLicenseAction.NAME,
+        DeleteLicenseAction.NAME,
+        // Autoscaling does not publish its actions to core, literal strings are needed.
+        "cluster:admin/autoscaling/put_autoscaling_policy",
+        "cluster:admin/autoscaling/delete_autoscaling_policy",
+        // Repository analysis actions are not mentioned in core, literal strings are needed.
+        "cluster:admin/repository/analyze",
+        "cluster:admin/repository/analyze/blob",
+        "cluster:admin/repository/analyze/blob/read",
+        "cluster:admin/repository/analyze/register",
+        // Node shutdown APIs are operator only
+        "cluster:admin/shutdown/create",
+        "cluster:admin/shutdown/get",
+        "cluster:admin/shutdown/delete",
+        // Node removal prevalidation API
+        PrevalidateNodeRemovalAction.NAME,
+        // Desired Nodes API
+        DeleteDesiredNodesAction.NAME,
+        GetDesiredNodesAction.NAME,
+        UpdateDesiredNodesAction.NAME,
+        GetDesiredBalanceAction.NAME,
+        DeleteDesiredBalanceAction.NAME
+    );
+
+    private final ClusterSettings clusterSettings;
+
+    public DefaultOperatorOnlyRegistry(ClusterSettings clusterSettings) {
+        this.clusterSettings = clusterSettings;
+    }
+
+    /**
+     * Check whether the given action and request qualify as operator-only. The method returns
+     * null if the action+request is NOT operator-only. Other it returns a violation object
+     * that contains the message for details.
+     */
+    public OperatorPrivilegesViolation check(String action, TransportRequest request) {
+        if (SIMPLE_ACTIONS.contains(action)) {
+            return () -> "action [" + action + "]";
+        } else if (ClusterUpdateSettingsAction.NAME.equals(action)) {
+            assert request instanceof ClusterUpdateSettingsRequest;
+            return checkClusterUpdateSettings((ClusterUpdateSettingsRequest) request);
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public OperatorPrivilegesViolation checkRest(RestHandler restHandler, RestRequest restRequest, RestChannel restChannel) {
+        return null; // no restrictions
+    }
+
+    private OperatorPrivilegesViolation checkClusterUpdateSettings(ClusterUpdateSettingsRequest request) {
+        List<String> operatorOnlySettingKeys = Stream.concat(
+            request.transientSettings().keySet().stream(),
+            request.persistentSettings().keySet().stream()
+        ).filter(k -> {
+            final Setting<?> setting = clusterSettings.get(k);
+            return setting != null && setting.isOperatorOnly();
+        }).toList();
+        if (false == operatorOnlySettingKeys.isEmpty()) {
+            return () -> (operatorOnlySettingKeys.size() == 1 ? "setting" : "settings")
+                + " ["
+                + Strings.collectionToDelimitedString(operatorOnlySettingKeys, ",")
+                + "]";
+        } else {
+            return null;
+        }
+    }
+
+}

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/operator/OperatorOnlyRegistry.java

@@ -7,98 +7,33 @@
 
 package org.elasticsearch.xpack.security.operator;
 
-import org.elasticsearch.action.admin.cluster.allocation.DeleteDesiredBalanceAction;
-import org.elasticsearch.action.admin.cluster.allocation.GetDesiredBalanceAction;
-import org.elasticsearch.action.admin.cluster.configuration.AddVotingConfigExclusionsAction;
-import org.elasticsearch.action.admin.cluster.configuration.ClearVotingConfigExclusionsAction;
-import org.elasticsearch.action.admin.cluster.desirednodes.DeleteDesiredNodesAction;
-import org.elasticsearch.action.admin.cluster.desirednodes.GetDesiredNodesAction;
-import org.elasticsearch.action.admin.cluster.desirednodes.UpdateDesiredNodesAction;
-import org.elasticsearch.action.admin.cluster.node.shutdown.PrevalidateNodeRemovalAction;
-import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsAction;
-import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
-import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.settings.ClusterSettings;
-import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.license.DeleteLicenseAction;
-import org.elasticsearch.license.PutLicenseAction;
+import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.transport.TransportRequest;
 
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Stream;
-
-public class OperatorOnlyRegistry {
-
-    public static final Set<String> SIMPLE_ACTIONS = Set.of(
-        AddVotingConfigExclusionsAction.NAME,
-        ClearVotingConfigExclusionsAction.NAME,
-        PutLicenseAction.NAME,
-        DeleteLicenseAction.NAME,
-        // Autoscaling does not publish its actions to core, literal strings are needed.
-        "cluster:admin/autoscaling/put_autoscaling_policy",
-        "cluster:admin/autoscaling/delete_autoscaling_policy",
-        // Repository analysis actions are not mentioned in core, literal strings are needed.
-        "cluster:admin/repository/analyze",
-        "cluster:admin/repository/analyze/blob",
-        "cluster:admin/repository/analyze/blob/read",
-        "cluster:admin/repository/analyze/register",
-        // Node shutdown APIs are operator only
-        "cluster:admin/shutdown/create",
-        "cluster:admin/shutdown/get",
-        "cluster:admin/shutdown/delete",
-        // Node removal prevalidation API
-        PrevalidateNodeRemovalAction.NAME,
-        // Desired Nodes API
-        DeleteDesiredNodesAction.NAME,
-        GetDesiredNodesAction.NAME,
-        UpdateDesiredNodesAction.NAME,
-        GetDesiredBalanceAction.NAME,
-        DeleteDesiredBalanceAction.NAME
-    );
-
-    private final ClusterSettings clusterSettings;
-
-    public OperatorOnlyRegistry(ClusterSettings clusterSettings) {
-        this.clusterSettings = clusterSettings;
-    }
+public interface OperatorOnlyRegistry {
 
     /**
      * Check whether the given action and request qualify as operator-only. The method returns
      * null if the action+request is NOT operator-only. Other it returns a violation object
      * that contains the message for details.
      */
-    public OperatorPrivilegesViolation check(String action, TransportRequest request) {
-        if (SIMPLE_ACTIONS.contains(action)) {
-            return () -> "action [" + action + "]";
-        } else if (ClusterUpdateSettingsAction.NAME.equals(action)) {
-            assert request instanceof ClusterUpdateSettingsRequest;
-            return checkClusterUpdateSettings((ClusterUpdateSettingsRequest) request);
-        } else {
-            return null;
-        }
-    }
+    OperatorPrivilegesViolation check(String action, TransportRequest request);
 
-    private OperatorPrivilegesViolation checkClusterUpdateSettings(ClusterUpdateSettingsRequest request) {
-        List<String> operatorOnlySettingKeys = Stream.concat(
-            request.transientSettings().keySet().stream(),
-            request.persistentSettings().keySet().stream()
-        ).filter(k -> {
-            final Setting<?> setting = clusterSettings.get(k);
-            return setting != null && setting.isOperatorOnly();
-        }).toList();
-        if (false == operatorOnlySettingKeys.isEmpty()) {
-            return () -> (operatorOnlySettingKeys.size() == 1 ? "setting" : "settings")
-                + " ["
-                + Strings.collectionToDelimitedString(operatorOnlySettingKeys, ",")
-                + "]";
-        } else {
-            return null;
-        }
-    }
+    /**
+     * Checks to see if a given {@link RestHandler} is subject to operator-only restrictions for the REST API. Any REST API may be
+     * fully or partially restricted. A fully restricted REST API mandates that the implementation call restChannel.sendResponse(...) and
+     * return a {@link OperatorPrivilegesViolation}. A partially restricted REST API mandates that the {@link RestRequest} is marked as
+     * restricted so that the downstream handler can behave appropriately. For example, to restrict the REST response the implementation
+     * should call {@link RestRequest#markResponseRestricted(String)} so that the downstream handler can properly restrict the response
+     * before returning to the client. Note - a partial restriction should return null.
+     * @param restHandler The {@link RestHandler} to check for any restrictions
+     * @param restRequest The {@link RestRequest} to check for any restrictions and mark any partially restricted REST API's
+     * @param restChannel The {@link RestChannel} to enforce fully restricted REST API's
+     * @return {@link OperatorPrivilegesViolation} iff the request was fully restricted and the response has been sent back to the client.
+     * else returns null.
+     */
+    OperatorPrivilegesViolation checkRest(RestHandler restHandler, RestRequest restRequest, RestChannel restChannel);
 
-    @FunctionalInterface
-    public interface OperatorPrivilegesViolation {
-        String message();
-    }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/operator/OperatorPrivileges.java

@@ -14,6 +14,9 @@ import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotR
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
@@ -49,6 +52,20 @@ public class OperatorPrivileges {
             ThreadContext threadContext
         );
 
+        /**
+         * Checks to see if a given {@link RestHandler} is subject to operator-only restrictions for the REST API. Any REST API may be
+         * fully or partially restricted. A fully restricted REST API mandates that the implementation results in
+         * restChannel.sendResponse(...) and return a {@code false} to prevent any further processing. A partially restricted REST API
+         * mandates that the {@link RestRequest} is marked as restricted and return {@code true}. No restrictions should also return
+         * {@code true}.
+         * @param restHandler The {@link RestHandler} to check for any restrictions
+         * @param restRequest The {@link RestRequest} to check for any restrictions and mark any partially restricted REST API's
+         * @param restChannel The {@link RestChannel} to enforce fully restricted REST API's
+         * @return {@code true} if processing the request should continue, {@code false} if processing the request should halt due to
+         * a fully restricted REST API
+         */
+        boolean checkRest(RestHandler restHandler, RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext);
+
         /**
          * When operator privileges are enabled, certain requests needs to be configured in a specific way
          * so that they respect operator only settings. For an example, the restore snapshot request
@@ -114,7 +131,7 @@ public class OperatorPrivileges {
             )) {
                 // Only check whether request is operator-only when user is NOT an operator
                 logger.trace("Checking operator-only violation for user [{}] and action [{}]", user, action);
-                final OperatorOnlyRegistry.OperatorPrivilegesViolation violation = operatorOnlyRegistry.check(action, request);
+                final OperatorPrivilegesViolation violation = operatorOnlyRegistry.check(action, request);
                 if (violation != null) {
                     return new ElasticsearchSecurityException("Operator privileges are required for " + violation.message());
                 }
@@ -122,6 +139,38 @@ public class OperatorPrivileges {
             return null;
         }
 
+        @Override
+        public boolean checkRest(RestHandler restHandler, RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {
+            if (false == shouldProcess()) {
+                return true;
+            }
+            if (false == AuthenticationField.PRIVILEGE_CATEGORY_VALUE_OPERATOR.equals(
+                threadContext.getHeader(AuthenticationField.PRIVILEGE_CATEGORY_KEY)
+            )) {
+                // Only check whether request is operator-only when user is NOT an operator
+                if (logger.isTraceEnabled()) {
+                    Authentication authentication = threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY);
+                    final User user = authentication.getEffectiveSubject().getUser();
+                    logger.trace("Checking for any operator-only REST violations for user [{}] and uri [{}]", user, restRequest.uri());
+                }
+                OperatorPrivilegesViolation violation = operatorOnlyRegistry.checkRest(restHandler, restRequest, restChannel);
+                if (violation != null) {
+                    if (logger.isDebugEnabled()) {
+                        Authentication authentication = threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY);
+                        final User user = authentication.getEffectiveSubject().getUser();
+                        logger.debug(
+                            "Found the following operator-only violation [{}] for user [{}] and uri [{}]",
+                            violation.message(),
+                            user,
+                            restRequest.uri()
+                        );
+                    }
+                    return false;
+                }
+            }
+            return true;
+        }
+
         public void maybeInterceptRequest(ThreadContext threadContext, TransportRequest request) {
             if (request instanceof RestoreSnapshotRequest) {
                 logger.debug("Intercepting [{}] for operator privileges", request);
@@ -132,6 +181,11 @@ public class OperatorPrivileges {
         private boolean shouldProcess() {
             return Security.OPERATOR_PRIVILEGES_FEATURE.check(licenseState);
         }
+
+        // for testing
+        public OperatorOnlyRegistry getOperatorOnlyRegistry() {
+            return operatorOnlyRegistry;
+        }
     }
 
     public static final OperatorPrivilegesService NOOP_OPERATOR_PRIVILEGES_SERVICE = new OperatorPrivilegesService() {
@@ -148,6 +202,11 @@ public class OperatorPrivileges {
             return null;
         }
 
+        @Override
+        public boolean checkRest(RestHandler restHandler, RestRequest restRequest, RestChannel restChannel, ThreadContext threadContext) {
+            return true;
+        }
+
         @Override
         public void maybeInterceptRequest(ThreadContext threadContext, TransportRequest request) {
             if (request instanceof RestoreSnapshotRequest) {

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/operator/OperatorPrivilegesViolation.java

@@ -0,0 +1,13 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.operator;
+
+@FunctionalInterface
+public interface OperatorPrivilegesViolation {
+    String message();
+}

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java

@@ -22,6 +22,7 @@ import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.xpack.security.audit.AuditTrailService;
 import org.elasticsearch.xpack.security.authc.support.SecondaryAuthenticator;
 import org.elasticsearch.xpack.security.authz.restriction.WorkflowService;
+import org.elasticsearch.xpack.security.operator.OperatorPrivileges;
 
 import java.util.List;
 
@@ -37,6 +38,7 @@ public class SecurityRestFilter implements RestHandler {
     private final boolean enabled;
     private final ThreadContext threadContext;
     private final WorkflowService workflowService;
+    private final OperatorPrivileges.OperatorPrivilegesService operatorPrivilegesService;
 
     public SecurityRestFilter(
         boolean enabled,
@@ -44,7 +46,8 @@ public class SecurityRestFilter implements RestHandler {
         SecondaryAuthenticator secondaryAuthenticator,
         AuditTrailService auditTrailService,
         WorkflowService workflowService,
-        RestHandler restHandler
+        RestHandler restHandler,
+        OperatorPrivileges.OperatorPrivilegesService operatorPrivilegesService
     ) {
         this.enabled = enabled;
         this.threadContext = threadContext;
@@ -52,6 +55,10 @@ public class SecurityRestFilter implements RestHandler {
         this.auditTrailService = auditTrailService;
         this.workflowService = workflowService;
         this.restHandler = restHandler;
+        // can be null if security is not enabled
+        this.operatorPrivilegesService = operatorPrivilegesService == null
+            ? OperatorPrivileges.NOOP_OPERATOR_PRIVILEGES_SERVICE
+            : operatorPrivilegesService;
     }
 
     @Override
@@ -94,11 +101,14 @@ public class SecurityRestFilter implements RestHandler {
 
     private void doHandleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
         threadContext.sanitizeHeaders();
-        try {
-            restHandler.handleRequest(request, channel, client);
-        } catch (Exception e) {
-            logger.debug(() -> format("Request handling failed for REST request [%s]", request.uri()), e);
-            throw e;
+        // operator privileges can short circuit to return a non-successful response
+        if (operatorPrivilegesService.checkRest(restHandler, request, channel, threadContext)) {
+            try {
+                restHandler.handleRequest(request, channel, client);
+            } catch (Exception e) {
+                logger.debug(() -> format("Request handling failed for REST request [%s]", request.uri()), e);
+                throw e;
+            }
         }
     }
 
@@ -133,6 +143,11 @@ public class SecurityRestFilter implements RestHandler {
         return restHandler.routes();
     }
 
+    // for testing
+    OperatorPrivileges.OperatorPrivilegesService getOperatorPrivilegesService() {
+        return operatorPrivilegesService;
+    }
+
     private RestRequest maybeWrapRestRequest(RestRequest restRequest) {
         if (restHandler instanceof RestRequestFilter) {
             return ((RestRequestFilter) restHandler).getFilteredRequest(restRequest);

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityTests.java

@@ -46,7 +46,10 @@ import org.elasticsearch.license.LicenseService;
 import org.elasticsearch.license.TestUtils;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.license.internal.XPackLicenseStatus;
+import org.elasticsearch.plugins.ExtensiblePlugin;
 import org.elasticsearch.plugins.MapperPlugin;
+import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.test.ESTestCase;
@@ -57,6 +60,7 @@ import org.elasticsearch.test.rest.FakeRestRequest;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.tracing.Tracer;
+import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.usage.UsageService;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.core.XPackField;
@@ -85,11 +89,16 @@ import org.elasticsearch.xpack.security.authc.Realms;
 import org.elasticsearch.xpack.security.authc.esnative.NativeUsersStore;
 import org.elasticsearch.xpack.security.authc.esnative.ReservedRealm;
 import org.elasticsearch.xpack.security.authc.service.CachingServiceAccountTokenStore;
+import org.elasticsearch.xpack.security.operator.DefaultOperatorOnlyRegistry;
+import org.elasticsearch.xpack.security.operator.OperatorOnlyRegistry;
+import org.elasticsearch.xpack.security.operator.OperatorPrivileges;
+import org.elasticsearch.xpack.security.operator.OperatorPrivilegesViolation;
 import org.hamcrest.Matchers;
 import org.junit.After;
 
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -107,6 +116,8 @@ import java.util.stream.Collectors;
 import static java.util.Collections.emptyMap;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_FORMAT_SETTING;
 import static org.elasticsearch.xpack.core.security.authc.RealmSettings.getFullSettingKey;
+import static org.elasticsearch.xpack.security.operator.OperatorPrivileges.NOOP_OPERATOR_PRIVILEGES_SERVICE;
+import static org.elasticsearch.xpack.security.operator.OperatorPrivileges.OPERATOR_PRIVILEGES_ENABLED;
 import static org.elasticsearch.xpack.security.support.SecuritySystemIndices.INTERNAL_MAIN_INDEX_FORMAT;
 import static org.elasticsearch.xpack.security.support.SecuritySystemIndices.SECURITY_MAIN_ALIAS;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -116,6 +127,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
@@ -143,22 +155,40 @@ public class SecurityTests extends ESTestCase {
         }
     }
 
-    private Collection<Object> createComponentsUtil(Settings settings, SecurityExtension... extensions) throws Exception {
+    public static class DummyOperatorOnlyRegistry implements OperatorOnlyRegistry {
+        @Override
+        public OperatorPrivilegesViolation check(String action, TransportRequest request) {
+            throw new RuntimeException("boom");
+        }
+
+        @Override
+        public OperatorPrivilegesViolation checkRest(RestHandler restHandler, RestRequest restRequest, RestChannel restChannel) {
+            throw new RuntimeException("boom");
+        }
+    }
+
+    private void constructNewSecurityObject(Settings settings, SecurityExtension... extensions) {
         Environment env = TestEnvironment.newEnvironment(settings);
-        NodeMetadata nodeMetadata = new NodeMetadata(randomAlphaOfLength(8), Version.CURRENT, Version.CURRENT);
         licenseState = new TestUtils.UpdatableLicenseState(settings);
         SSLService sslService = new SSLService(env);
-        security = new Security(settings, Arrays.asList(extensions)) {
-            @Override
-            protected XPackLicenseState getLicenseState() {
-                return licenseState;
-            }
+        if (security == null) {
+            security = new Security(settings, Arrays.asList(extensions)) {
+                @Override
+                protected XPackLicenseState getLicenseState() {
+                    return licenseState;
+                }
+
+                @Override
+                protected SSLService getSslService() {
+                    return sslService;
+                }
+            };
+        }
+    }
 
-            @Override
-            protected SSLService getSslService() {
-                return sslService;
-            }
-        };
+    private Collection<Object> createComponentsUtil(Settings settings) throws Exception {
+        Environment env = TestEnvironment.newEnvironment(settings);
+        NodeMetadata nodeMetadata = new NodeMetadata(randomAlphaOfLength(8), Version.CURRENT, Version.CURRENT);
         ThreadPool threadPool = mock(ThreadPool.class);
         ClusterService clusterService = mock(ClusterService.class);
         settings = Security.additionalSettings(settings, true);
@@ -195,7 +225,8 @@ public class SecurityTests extends ESTestCase {
             .put(testSettings)
             .put("path.home", createTempDir())
             .build();
-        return createComponentsUtil(settings, extensions);
+        constructNewSecurityObject(settings, extensions);
+        return createComponentsUtil(settings);
     }
 
     private static <T> T findComponent(Class<T> type, Collection<Object> components) {
@@ -897,6 +928,112 @@ public class SecurityTests extends ESTestCase {
         );
     }
 
+    public void testLoadExtensions() throws Exception {
+        Settings settings = Settings.builder()
+            .put("xpack.security.enabled", true)
+            .put("path.home", createTempDir())
+            .put(OPERATOR_PRIVILEGES_ENABLED.getKey(), true)
+            .build();
+        constructNewSecurityObject(settings);
+        security.loadExtensions(new ExtensiblePlugin.ExtensionLoader() {
+            @Override
+            @SuppressWarnings("unchecked")
+            public <T> List<T> loadExtensions(Class<T> extensionPointType) {
+                List<Object> extensions = new ArrayList<>();
+                if (extensionPointType == OperatorOnlyRegistry.class) {
+                    extensions.add(new DummyOperatorOnlyRegistry());
+                }
+                return (List<T>) extensions;
+            }
+        });
+        createComponentsUtil(settings);
+        OperatorPrivileges.OperatorPrivilegesService operatorPrivilegesService = security.getOperatorPrivilegesService();
+        assertThat(operatorPrivilegesService, instanceOf(OperatorPrivileges.DefaultOperatorPrivilegesService.class));
+        OperatorOnlyRegistry registry = ((OperatorPrivileges.DefaultOperatorPrivilegesService) operatorPrivilegesService)
+            .getOperatorOnlyRegistry();
+        assertThat(registry, instanceOf(DummyOperatorOnlyRegistry.class));
+    }
+
+    public void testLoadNoExtensions() throws Exception {
+        Settings settings = Settings.builder()
+            .put("xpack.security.enabled", true)
+            .put("path.home", createTempDir())
+            .put(OPERATOR_PRIVILEGES_ENABLED.getKey(), true)
+            .build();
+        constructNewSecurityObject(settings);
+        security.loadExtensions(new ExtensiblePlugin.ExtensionLoader() {
+            @Override
+            public <T> List<T> loadExtensions(Class<T> extensionPointType) {
+                return new ArrayList<>();
+            }
+        });
+        createComponentsUtil(settings);
+        OperatorPrivileges.OperatorPrivilegesService operatorPrivilegesService = security.getOperatorPrivilegesService();
+        assertThat(operatorPrivilegesService, instanceOf(OperatorPrivileges.DefaultOperatorPrivilegesService.class));
+        OperatorOnlyRegistry registry = ((OperatorPrivileges.DefaultOperatorPrivilegesService) operatorPrivilegesService)
+            .getOperatorOnlyRegistry();
+        assertThat(registry, instanceOf(DefaultOperatorOnlyRegistry.class));
+
+    }
+
+    public void testLoadExtensionsWhenOperatorPrivsAreDisabled() throws Exception {
+        assumeFalse("feature flag for serverless is expected to be false", DiscoveryNode.isServerless());
+        Settings.Builder settingsBuilder = Settings.builder().put("xpack.security.enabled", true).put("path.home", createTempDir());
+
+        if (randomBoolean()) {
+            settingsBuilder.put(OPERATOR_PRIVILEGES_ENABLED.getKey(), false); // doesn't matter if explicit or implicitly disabled
+        }
+
+        Settings settings = settingsBuilder.build();
+        constructNewSecurityObject(settings);
+        security.loadExtensions(new ExtensiblePlugin.ExtensionLoader() {
+            @Override
+            @SuppressWarnings("unchecked")
+            public <T> List<T> loadExtensions(Class<T> extensionPointType) {
+                List<Object> extensions = new ArrayList<>();
+                if (extensionPointType == OperatorOnlyRegistry.class) {
+                    if (randomBoolean()) {
+                        extensions.add(new DummyOperatorOnlyRegistry()); // won't ever be used
+                    }
+                }
+                return (List<T>) extensions;
+            }
+        });
+        createComponentsUtil(settings);
+        OperatorPrivileges.OperatorPrivilegesService operatorPrivilegesService = security.getOperatorPrivilegesService();
+        assertThat(operatorPrivilegesService, is(NOOP_OPERATOR_PRIVILEGES_SERVICE));
+    }
+
+    public void testLoadExtensionsWhenOperatorPrivsAreDisabledAndServerless() throws Exception {
+        assumeTrue("feature flag for serverless is expected to be true", DiscoveryNode.isServerless());
+        Settings.Builder settingsBuilder = Settings.builder().put("xpack.security.enabled", true).put("path.home", createTempDir());
+
+        if (randomBoolean()) {
+            settingsBuilder.put(OPERATOR_PRIVILEGES_ENABLED.getKey(), false); // doesn't matter if explicit or implicitly disabled
+        }
+
+        Settings settings = settingsBuilder.build();
+        constructNewSecurityObject(settings);
+        security.loadExtensions(new ExtensiblePlugin.ExtensionLoader() {
+            @Override
+            @SuppressWarnings("unchecked")
+            public <T> List<T> loadExtensions(Class<T> extensionPointType) {
+                List<Object> extensions = new ArrayList<>();
+                if (extensionPointType == OperatorOnlyRegistry.class) {
+                    if (randomBoolean()) {
+                        extensions.add(new DummyOperatorOnlyRegistry()); // will be used
+                    }
+                }
+                return (List<T>) extensions;
+            }
+        });
+        createComponentsUtil(settings);
+        OperatorPrivileges.OperatorPrivilegesService operatorPrivilegesService = security.getOperatorPrivilegesService();
+        OperatorOnlyRegistry registry = ((OperatorPrivileges.DefaultOperatorPrivilegesService) operatorPrivilegesService)
+            .getOperatorOnlyRegistry();
+        assertThat(registry, instanceOf(DummyOperatorOnlyRegistry.class));
+    }
+
     private void verifyHasAuthenticationHeaderValue(Exception e, String... expectedValues) {
         assertThat(e, instanceOf(ElasticsearchSecurityException.class));
         assertThat(((ElasticsearchSecurityException) e).getHeader("WWW-Authenticate"), notNullValue());

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/operator/OperatorOnlyRegistryTests.java → x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/operator/DefaultOperatorOnlyRegistryTests.java

@@ -32,7 +32,7 @@ import static org.hamcrest.Matchers.containsString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-public class OperatorOnlyRegistryTests extends ESTestCase {
+public class DefaultOperatorOnlyRegistryTests extends ESTestCase {
 
     private static final Set<Setting<?>> DYNAMIC_SETTINGS = ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.stream()
         .filter(Setting::isDynamic)
@@ -49,18 +49,18 @@ public class OperatorOnlyRegistryTests extends ESTestCase {
         HTTP_FILTER_DENY_SETTING
     );
 
-    private OperatorOnlyRegistry operatorOnlyRegistry;
+    private DefaultOperatorOnlyRegistry operatorOnlyRegistry;
 
     @Before
     public void init() {
         final Set<Setting<?>> settingsSet = new HashSet<>(IP_FILTER_SETTINGS);
         settingsSet.addAll(DYNAMIC_SETTINGS);
-        operatorOnlyRegistry = new OperatorOnlyRegistry(new ClusterSettings(Settings.EMPTY, settingsSet));
+        operatorOnlyRegistry = new DefaultOperatorOnlyRegistry(new ClusterSettings(Settings.EMPTY, settingsSet));
     }
 
     public void testSimpleOperatorOnlyApi() {
-        for (final String actionName : OperatorOnlyRegistry.SIMPLE_ACTIONS) {
-            final OperatorOnlyRegistry.OperatorPrivilegesViolation violation = operatorOnlyRegistry.check(actionName, null);
+        for (final String actionName : DefaultOperatorOnlyRegistry.SIMPLE_ACTIONS) {
+            final OperatorPrivilegesViolation violation = operatorOnlyRegistry.check(actionName, null);
             assertNotNull(violation);
             assertThat(violation.message(), containsString("action [" + actionName + "]"));
         }
@@ -68,7 +68,7 @@ public class OperatorOnlyRegistryTests extends ESTestCase {
 
     public void testNonOperatorOnlyApi() {
         final String actionName = randomValueOtherThanMany(
-            OperatorOnlyRegistry.SIMPLE_ACTIONS::contains,
+            DefaultOperatorOnlyRegistry.SIMPLE_ACTIONS::contains,
             () -> randomAlphaOfLengthBetween(10, 40)
         );
         assertNull(operatorOnlyRegistry.check(actionName, null));
@@ -78,7 +78,7 @@ public class OperatorOnlyRegistryTests extends ESTestCase {
         final ClusterUpdateSettingsRequest request;
         final Setting<?> transientSetting;
         final Setting<?> persistentSetting;
-        final OperatorOnlyRegistry.OperatorPrivilegesViolation violation;
+        final OperatorPrivilegesViolation violation;
 
         switch (randomIntBetween(0, 3)) {
             case 0 -> {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/operator/OperatorPrivilegesTests.java → x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/operator/DefaultOperatorPrivilegesTests.java

@@ -16,6 +16,9 @@ import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.license.MockLicenseState;
+import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.MockLogAppender;
 import org.elasticsearch.transport.TransportRequest;
@@ -40,21 +43,22 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
-public class OperatorPrivilegesTests extends ESTestCase {
+public class DefaultOperatorPrivilegesTests extends ESTestCase {
 
     private MockLicenseState xPackLicenseState;
     private FileOperatorUsersStore fileOperatorUsersStore;
-    private OperatorOnlyRegistry operatorOnlyRegistry;
+    private DefaultOperatorOnlyRegistry operatorOnlyRegistry;
     private OperatorPrivilegesService operatorPrivilegesService;
 
     @Before
     public void init() {
         xPackLicenseState = mock(MockLicenseState.class);
         fileOperatorUsersStore = mock(FileOperatorUsersStore.class);
-        operatorOnlyRegistry = mock(OperatorOnlyRegistry.class);
+        operatorOnlyRegistry = mock(DefaultOperatorOnlyRegistry.class);
         operatorPrivilegesService = new DefaultOperatorPrivilegesService(xPackLicenseState, fileOperatorUsersStore, operatorOnlyRegistry);
     }
 
@@ -265,4 +269,22 @@ public class OperatorPrivilegesTests extends ESTestCase {
         NOOP_OPERATOR_PRIVILEGES_SERVICE.maybeInterceptRequest(threadContext, mock(TransportRequest.class));
     }
 
+    public void testCheckRest() {
+        final Settings settings = Settings.builder().put("xpack.security.operator_privileges.enabled", true).build();
+        when(xPackLicenseState.isAllowed(Security.OPERATOR_PRIVILEGES_FEATURE)).thenReturn(true);
+        RestHandler restHandler = mock(RestHandler.class);
+        RestRequest restRequest = mock(RestRequest.class);
+        RestChannel restChannel = mock(RestChannel.class);
+        ThreadContext threadContext = new ThreadContext(settings);
+
+        // not an operator
+        when(operatorOnlyRegistry.checkRest(restHandler, restRequest, restChannel)).thenReturn(() -> "violation!");
+        assertFalse(operatorPrivilegesService.checkRest(restHandler, restRequest, restChannel, threadContext));
+        Mockito.clearInvocations(operatorOnlyRegistry);
+
+        // is an operator
+        threadContext.putHeader(AuthenticationField.PRIVILEGE_CATEGORY_KEY, AuthenticationField.PRIVILEGE_CATEGORY_VALUE_OPERATOR);
+        verifyNoInteractions(operatorOnlyRegistry);
+        assertTrue(operatorPrivilegesService.checkRest(restHandler, restRequest, restChannel, threadContext));
+    }
 }

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/rest/SecurityRestFilterTests.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.security.rest;
 import com.nimbusds.jose.util.StandardCharset;
 
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesArray;
@@ -27,6 +28,7 @@ import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.SecuritySettingsSourceField;
 import org.elasticsearch.test.rest.FakeRestRequest;
+import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xcontent.DeprecationHandler;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentType;
@@ -44,6 +46,7 @@ import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authc.support.SecondaryAuthenticator;
 import org.elasticsearch.xpack.security.authz.restriction.WorkflowService;
 import org.elasticsearch.xpack.security.authz.restriction.WorkflowServiceTests.TestBaseRestHandler;
+import org.elasticsearch.xpack.security.operator.OperatorPrivileges;
 import org.junit.Before;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
@@ -56,6 +59,7 @@ import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.test.ActionListenerUtils.anyActionListener;
+import static org.elasticsearch.xpack.security.operator.OperatorPrivileges.NOOP_OPERATOR_PRIVILEGES_SERVICE;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItem;
@@ -68,6 +72,7 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
@@ -88,13 +93,18 @@ public class SecurityRestFilterTests extends ESTestCase {
         restHandler = mock(RestHandler.class);
         threadContext = new ThreadContext(Settings.EMPTY);
         secondaryAuthenticator = new SecondaryAuthenticator(Settings.EMPTY, threadContext, authcService, new AuditTrailService(null, null));
-        filter = new SecurityRestFilter(
+        filter = getFilter(NOOP_OPERATOR_PRIVILEGES_SERVICE);
+    }
+
+    private SecurityRestFilter getFilter(OperatorPrivileges.OperatorPrivilegesService privilegesService) {
+        return new SecurityRestFilter(
             true,
             threadContext,
             secondaryAuthenticator,
             new AuditTrailService(null, null),
             new WorkflowService(),
-            restHandler
+            restHandler,
+            privilegesService
         );
     }
 
@@ -167,8 +177,10 @@ public class SecurityRestFilterTests extends ESTestCase {
             secondaryAuthenticator,
             mock(AuditTrailService.class),
             mock(WorkflowService.class),
-            restHandler
+            restHandler,
+            null
         );
+        assertEquals(NOOP_OPERATOR_PRIVILEGES_SERVICE, filter.getOperatorPrivilegesService());
         RestRequest request = mock(RestRequest.class);
         filter.handleRequest(request, channel, null);
         verify(restHandler).handleRequest(request, channel, null);
@@ -220,7 +232,8 @@ public class SecurityRestFilterTests extends ESTestCase {
             secondaryAuthenticator,
             new AuditTrailService(auditTrail, licenseState),
             new WorkflowService(),
-            restHandler
+            restHandler,
+            NOOP_OPERATOR_PRIVILEGES_SERVICE
         );
 
         filter.handleRequest(restRequest, channel, null);
@@ -291,7 +304,8 @@ public class SecurityRestFilterTests extends ESTestCase {
             secondaryAuthenticator,
             new AuditTrailService(null, null),
             workflowService,
-            restHandler
+            restHandler,
+            null
         );
 
         RestRequest request = mock(RestRequest.class);
@@ -317,7 +331,8 @@ public class SecurityRestFilterTests extends ESTestCase {
             secondaryAuthenticator,
             new AuditTrailService(null, null),
             workflowService,
-            restHandler
+            restHandler,
+            null
         );
 
         RestRequest request = mock(RestRequest.class);
@@ -325,5 +340,47 @@ public class SecurityRestFilterTests extends ESTestCase {
         assertThat(workflowService.readWorkflowFromThreadContext(threadContext), nullValue());
     }
 
+    public void testCheckRest() throws Exception {
+        for (Boolean isOperator : new Boolean[] { Boolean.TRUE, Boolean.FALSE }) {
+            RestRequest request = mock(RestRequest.class);
+            try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
+                SecurityRestFilter filter = getFilter(new OperatorPrivileges.OperatorPrivilegesService() {
+                    @Override
+                    public void maybeMarkOperatorUser(Authentication authentication, ThreadContext threadContext) {}
+
+                    @Override
+                    public ElasticsearchSecurityException check(
+                        Authentication authentication,
+                        String action,
+                        TransportRequest request,
+                        ThreadContext threadContext
+                    ) {
+                        return null;
+                    }
+
+                    @Override
+                    public boolean checkRest(
+                        RestHandler restHandler,
+                        RestRequest restRequest,
+                        RestChannel restChannel,
+                        ThreadContext threadContext
+                    ) {
+                        return isOperator;
+                    }
+
+                    @Override
+                    public void maybeInterceptRequest(ThreadContext threadContext, TransportRequest request) {}
+                });
+
+                filter.handleRequest(request, channel, null);
+                if (isOperator) {
+                    verify(restHandler).handleRequest(request, channel, null);
+                } else {
+                    verify(restHandler, never()).handleRequest(request, channel, null);
+                }
+            }
+        }
+    }
+
     private interface FilteredRestHandler extends RestHandler, RestRequestFilter {}
 }