Marking operator user no longer depends on license state (#79595)

When marking and checking both depend on licene state, there is a
potential and rare racing condition where license is incompatible at
marking time but becomes compatible at checking time. The result is an
unauthorized error. Since enforcement at checking time is the true value
of the feature of operator privileges, this PR removes the license check
at marking time which in turns eliminate the possible racing.

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

@@ -627,6 +627,7 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
         final OperatorPrivilegesService operatorPrivilegesService;
         final boolean operatorPrivilegesEnabled = OPERATOR_PRIVILEGES_ENABLED.get(settings);
         if (operatorPrivilegesEnabled) {
+            logger.info("operator privileges are enabled");
             operatorPrivilegesService = new OperatorPrivileges.DefaultOperatorPrivilegesService(getLicenseState(),
                 new FileOperatorUsersStore(environment, resourceWatcherService),
                 new OperatorOnlyRegistry(clusterService.getClusterSettings()));

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

@@ -66,9 +66,7 @@ public class OperatorPrivileges {
         }
 
         public void maybeMarkOperatorUser(Authentication authentication, ThreadContext threadContext) {
-            if (false == shouldProcess()) {
-                return;
-            }
+            // Always mark the thread context for operator users regardless of license state which is enforced at check time
             final User user = authentication.getUser();
             // Let internal users pass, they are exempt from marking and checking
             // Also check run_as, it is impossible to run_as internal users, but just to be extra safe

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

@@ -35,11 +35,13 @@ import org.mockito.Mockito;
 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.notNullValue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
 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.verifyZeroInteractions;
 import static org.mockito.Mockito.when;
@@ -59,16 +61,20 @@ public class OperatorPrivilegesTests extends ESTestCase {
         operatorPrivilegesService = new DefaultOperatorPrivilegesService(xPackLicenseState, fileOperatorUsersStore, operatorOnlyRegistry);
     }
 
-    public void testWillNotProcessWhenFeatureIsDisabledOrLicenseDoesNotSupport() {
-        final Settings settings = Settings.builder()
-            .put("xpack.security.operator_privileges.enabled", randomBoolean())
-            .build();
-        when(xPackLicenseState.isAllowed(Security.OPERATOR_PRIVILEGES_FEATURE)).thenReturn(false);
-        final ThreadContext threadContext = new ThreadContext(settings);
+    public void testWillMarkThreadContextForAllLicenses() {
+        when(xPackLicenseState.isAllowed(Security.OPERATOR_PRIVILEGES_FEATURE)).thenReturn(randomBoolean());
 
-        operatorPrivilegesService.maybeMarkOperatorUser(mock(Authentication.class), threadContext);
-        verifyZeroInteractions(fileOperatorUsersStore);
+        final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        final Authentication authentication = mock(Authentication.class);
+        when(authentication.getUser()).thenReturn(new User(randomAlphaOfLengthBetween(3, 8)));
+        operatorPrivilegesService.maybeMarkOperatorUser(authentication, threadContext);
+        verify(fileOperatorUsersStore, times(1)).isOperatorUser(authentication);
+        assertThat(threadContext.getHeader(AuthenticationField.PRIVILEGE_CATEGORY_KEY), notNullValue());
+    }
 
+    public void testWillNotCheckWhenLicenseDoesNotSupport() {
+        when(xPackLicenseState.isAllowed(Security.OPERATOR_PRIVILEGES_FEATURE)).thenReturn(false);
+        final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
         final ElasticsearchSecurityException e =
             operatorPrivilegesService.check(mock(Authentication.class), "cluster:action", mock(TransportRequest.class), threadContext);
         assertNull(e);