Fix unique realm name check to cover default realms (#87999)

All enabled realms must have unique names. This PR tightened the logic
for ensuring realm name uniqueness. Previously the unique name check
does not cover realms that are automatically enabled.

Relates: #69096

Author: ywangd

docs/changelog/87999.yaml

@@ -0,0 +1,5 @@
+pr: 87999
+summary: Fix unique realm name check to cover default realms
+area: Authentication
+type: bug
+issues: []

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

@@ -272,18 +272,11 @@ protected List<Realm> initRealms(List<RealmConfig> realmConfigs) throws Exceptio
 
         checkUniqueOrders(orderToRealmName);
         Collections.sort(realms);
+        ensureUniqueExplicitlyConfiguredRealmNames(nameToRealmIdentifier);
 
         maybeAddBasicRealms(realms, realmConfigs);
         // always add built in first!
-        realms.add(0, reservedRealm);
-        String duplicateRealms = nameToRealmIdentifier.entrySet()
-            .stream()
-            .filter(entry -> entry.getValue().size() > 1)
-            .map(entry -> entry.getKey() + ": " + entry.getValue())
-            .collect(Collectors.joining("; "));
-        if (Strings.hasText(duplicateRealms)) {
-            throw new IllegalArgumentException("Found multiple realms configured with the same name: " + duplicateRealms + "");
-        }
+        addReservedRealm(realms);
         logDeprecationForReservedPrefixedRealmNames(reservedPrefixedRealmIdentifiers);
         return Collections.unmodifiableList(realms);
     }
@@ -372,8 +365,9 @@ protected void doClose() throws IOException {
     private void maybeAddBasicRealms(List<Realm> realms, List<RealmConfig> realmConfigs) throws Exception {
         final Set<String> disabledBasicRealmTypes = findDisabledBasicRealmTypes(realmConfigs);
         final Set<String> realmTypes = realms.stream().map(Realm::type).collect(Collectors.toUnmodifiableSet());
-        // Add native realm first so that file realm will be in the beginning
+        // Add native realm first so that file realm will be added before it
         if (false == disabledBasicRealmTypes.contains(NativeRealmSettings.TYPE) && false == realmTypes.contains(NativeRealmSettings.TYPE)) {
+            ensureRealmNameIsAvailable(realms, NativeRealmSettings.DEFAULT_NAME);
             var nativeRealmId = new RealmConfig.RealmIdentifier(NativeRealmSettings.TYPE, NativeRealmSettings.DEFAULT_NAME);
             var realmConfig = new RealmConfig(
                 nativeRealmId,
@@ -385,6 +379,7 @@ private void maybeAddBasicRealms(List<Realm> realms, List<RealmConfig> realmConf
             realms.add(0, factories.get(NativeRealmSettings.TYPE).create(realmConfig));
         }
         if (false == disabledBasicRealmTypes.contains(FileRealmSettings.TYPE) && false == realmTypes.contains(FileRealmSettings.TYPE)) {
+            ensureRealmNameIsAvailable(realms, FileRealmSettings.DEFAULT_NAME);
             var fileRealmId = new RealmConfig.RealmIdentifier(FileRealmSettings.TYPE, FileRealmSettings.DEFAULT_NAME);
             var realmConfig = new RealmConfig(
                 fileRealmId,
@@ -397,6 +392,29 @@ private void maybeAddBasicRealms(List<Realm> realms, List<RealmConfig> realmConf
         }
     }
 
+    private void addReservedRealm(List<Realm> realms) {
+        ensureRealmNameIsAvailable(realms, ReservedRealm.NAME);
+        realms.add(0, reservedRealm);
+    }
+
+    /**
+     * Check that the given realmName is not yet used by the given list of realms.
+     */
+    private void ensureRealmNameIsAvailable(List<Realm> realms, String realmName) {
+        assert realms.size() == realms.stream().map(Realm::name).collect(Collectors.toUnmodifiableSet()).size()
+            : "existing realm names must be unique";
+        final Realm misNamedRealm = realms.stream().filter(realm -> realmName.equals(realm.name())).findFirst().orElse(null);
+        if (misNamedRealm != null) {
+            throw new IllegalArgumentException(
+                "Found realm configured with name clashing with the ["
+                    + realmName
+                    + "] realm: ["
+                    + (RealmSettings.realmSettingPrefix(misNamedRealm.type()) + misNamedRealm.name())
+                    + "]"
+            );
+        }
+    }
+
     private static Settings ensureOrderSetting(Settings settings, RealmConfig.RealmIdentifier realmIdentifier, int order) {
         String orderSettingKey = RealmSettings.realmSettingPrefix(realmIdentifier) + "order";
         return Settings.builder().put(settings).put(orderSettingKey, order).build();
@@ -413,6 +431,17 @@ private static void checkUniqueOrders(Map<Integer, Set<String>> orderToRealmName
         }
     }
 
+    private void ensureUniqueExplicitlyConfiguredRealmNames(Map<String, Set<String>> nameToRealmIdentifier) {
+        String duplicateRealms = nameToRealmIdentifier.entrySet()
+            .stream()
+            .filter(entry -> entry.getValue().size() > 1)
+            .map(entry -> entry.getKey() + ": " + entry.getValue())
+            .collect(Collectors.joining("; "));
+        if (Strings.hasText(duplicateRealms)) {
+            throw new IllegalArgumentException("Found multiple realms configured with the same name: " + duplicateRealms + "");
+        }
+    }
+
     private List<RealmConfig> buildRealmConfigs() {
         final Map<RealmConfig.RealmIdentifier, Settings> realmsSettings = RealmSettings.getRealmSettings(settings);
         final Set<String> internalTypes = new HashSet<>();

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

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.license.License;
@@ -125,16 +126,7 @@ public void init() throws Exception {
 
         threadContext = new ThreadContext(Settings.EMPTY);
 
-        ThreadPool threadPool = mock(ThreadPool.class);
-        when(threadPool.getThreadContext()).thenReturn(threadContext);
-
-        reservedRealm = new ReservedRealm(
-            mock(Environment.class),
-            Settings.EMPTY,
-            mock(NativeUsersStore.class),
-            new AnonymousUser(Settings.EMPTY),
-            threadPool
-        );
+        reservedRealm = buildReservedRealm();
         allowAllRealms();
     }
 
@@ -408,27 +400,6 @@ public void testRealmAssignedToMultipleDomainsNotPermitted() {
         );
     }
 
-    public void testReservedRealmIsAlwaysDomainless() throws Exception {
-        int realmId = randomIntBetween(0, randomRealmTypesCount - 1);
-        Settings settings = Settings.builder()
-            .put("xpack.security.authc.realms.type_" + realmId + ".reserved.order", 2)
-            .put("xpack.security.authc.domains.domain_reserved.realms", "reserved")
-            .put("xpack.security.authc.realms." + NativeRealmSettings.TYPE + ".disabled_native.enabled", false)
-            .put("xpack.security.authc.realms." + FileRealmSettings.TYPE + ".disabled_file.enabled", false)
-            .put("path.home", createTempDir())
-            .build();
-        Environment env = TestEnvironment.newEnvironment(settings);
-        Realms realms = new Realms(settings, env, factories, licenseState, threadContext, reservedRealm);
-        Iterator<Realm> iterator = realms.iterator();
-        assertThat(iterator.hasNext(), is(true));
-        Realm realm = iterator.next();
-        assertThat(realm, is(reservedRealm));
-        realm = iterator.next();
-        assertThat(realm.type(), is("type_" + realmId));
-        assertThat(realm.name(), is("reserved"));
-        assertThat(realm.realmRef().getDomain().name(), is("domain_reserved"));
-    }
-
     public void testDomainWithUndefinedRealms() {
         Settings settings = Settings.builder()
             .put("xpack.security.authc.domains.test_domain.realms", "reserved, default_file")
@@ -1097,6 +1068,147 @@ public void testWarningsForReservedPrefixedRealmNames() throws Exception {
         );
     }
 
+    public void testExplicitlyConfiguredRealmNamesMustNotClashWithDefaultRealmNames() throws Exception {
+        final Tuple<String, String> defaultRealm = randomFrom(
+            new Tuple<>("reserved", "reserved"),
+            new Tuple<>("file", "default_file"),
+            new Tuple<>("native", "default_native")
+        );
+
+        final String configuredRealmType = randomValueOtherThan(defaultRealm.v1(), () -> randomAlphaOfLengthBetween(3, 18));
+        factories.put(configuredRealmType, DummyRealm::new);
+        final String realmSettingKey = "xpack.security.authc.realms." + configuredRealmType + "." + defaultRealm.v2();
+
+        // The name of an explicitly configured realm must not be either default_native, default_file or "reserved"
+        final Settings settings1 = Settings.builder()
+            .put("path.home", createTempDir())
+            .put(realmSettingKey + ".order", randomInt())
+            .build();
+        final IllegalArgumentException e1 = expectThrows(
+            IllegalArgumentException.class,
+            () -> new Realms(
+                settings1,
+                TestEnvironment.newEnvironment(settings1),
+                factories,
+                licenseState,
+                threadContext,
+                buildReservedRealm()
+            )
+        );
+        assertThat(
+            e1.getMessage(),
+            containsString(
+                "Found realm configured with name clashing with the [" + defaultRealm.v2() + "] realm: [" + realmSettingKey + "]"
+            )
+        );
+
+        // It is OK if the explicitly configured realm is disabled
+        final Settings settings2 = Settings.builder()
+            .put("path.home", createTempDir())
+            .put(realmSettingKey + ".order", randomInt())
+            .put(realmSettingKey + ".enabled", false)
+            .build();
+        final Realms realms2 = new Realms(
+            settings2,
+            TestEnvironment.newEnvironment(settings2),
+            factories,
+            licenseState,
+            threadContext,
+            buildReservedRealm()
+        );
+        assertThat(realms2.getActiveRealms().stream().map(Realm::type).toList(), containsInAnyOrder("reserved", "file", "native"));
+
+        // This block of tests do not apply to reserved realm since it is not configurable
+        if ("file".equals(defaultRealm.v1()) || "native".equals(defaultRealm.v1())) {
+            final String autoEnabledRealmType = "file".equals(defaultRealm.v1()) ? "native" : "file";
+
+            // It is OK if the default file or native realm is disabled
+            final Settings settings3 = Settings.builder()
+                .put("path.home", createTempDir())
+                .put(realmSettingKey + ".order", randomInt())
+                .put("xpack.security.authc.realms." + defaultRealm.v1() + "." + randomAlphaOfLengthBetween(3, 18) + ".enabled", false)
+                .build();
+            final Realms realms3 = new Realms(
+                settings3,
+                TestEnvironment.newEnvironment(settings3),
+                factories,
+                licenseState,
+                threadContext,
+                buildReservedRealm()
+            );
+            assertThat(
+                realms3.getActiveRealms().stream().map(Realm::type).toList(),
+                containsInAnyOrder("reserved", configuredRealmType, autoEnabledRealmType)
+            );
+
+            // It is OK if file or native realm is named something else
+            final Settings settings4 = Settings.builder()
+                .put("path.home", createTempDir())
+                .put(realmSettingKey + ".order", randomInt())
+                .put("xpack.security.authc.realms." + defaultRealm.v1() + "." + randomAlphaOfLengthBetween(3, 18) + ".order", randomInt())
+                .build();
+            final Realms realms4 = new Realms(
+                settings4,
+                TestEnvironment.newEnvironment(settings4),
+                factories,
+                licenseState,
+                threadContext,
+                buildReservedRealm()
+            );
+            assertThat(
+                realms4.getActiveRealms().stream().map(Realm::type).toList(),
+                containsInAnyOrder("reserved", configuredRealmType, "file", "native")
+            );
+
+            // All explicitly configured realm names still must be unique
+            final Settings settings5 = Settings.builder()
+                .put("path.home", createTempDir())
+                .put(realmSettingKey + ".order", randomInt())
+                .put("xpack.security.authc.realms." + defaultRealm.v1() + "." + defaultRealm.v2() + ".order", randomInt())
+                .build();
+            final IllegalArgumentException e5 = expectThrows(
+                IllegalArgumentException.class,
+                () -> new Realms(
+                    settings5,
+                    TestEnvironment.newEnvironment(settings5),
+                    factories,
+                    licenseState,
+                    threadContext,
+                    buildReservedRealm()
+                )
+            );
+            assertThat(e5.getMessage(), containsString("Found multiple realms configured with the same name"));
+        }
+
+        // It is OK to explicitly configure file to be default_file and native to be default_native
+        final Settings settings6 = Settings.builder()
+            .put("path.home", createTempDir())
+            .put("xpack.security.authc.realms.file.default_file.order", randomInt())
+            .put("xpack.security.authc.realms.native.default_native.order", randomInt())
+            .build();
+        final Realms realms6 = new Realms(
+            settings6,
+            TestEnvironment.newEnvironment(settings6),
+            factories,
+            licenseState,
+            threadContext,
+            buildReservedRealm()
+        );
+        assertThat(realms6.getActiveRealms().stream().map(Realm::type).toList(), containsInAnyOrder("reserved", "file", "native"));
+    }
+
+    private ReservedRealm buildReservedRealm() {
+        final ThreadPool threadPool = mock(ThreadPool.class);
+        when(threadPool.getThreadContext()).thenReturn(threadContext);
+        return new ReservedRealm(
+            mock(Environment.class),
+            Settings.EMPTY,
+            mock(NativeUsersStore.class),
+            new AnonymousUser(Settings.EMPTY),
+            threadPool
+        );
+    }
+
     private boolean randomDisableRealm(Settings.Builder builder, String type) {
         final boolean disabled = randomBoolean();
         if (disabled) {