Enforce realm name uniqueness (#46580)

We depend on file realms being unique in a number of places. Pre
7.0 this was enforced by the fact that the multiple realm types
with different name would mean identical configuration keys and
cause configuration parsing errors. Since we intoduced affix
settings for realms this is not the case any more as the realm type
is part of the configuration key.
This change adds a check when building realms which will explicitly
fail if multiple realms are defined with the same name.

Backport of #46253

Author: jkakavas

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

@@ -8,6 +8,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.CountDown;
@@ -184,6 +185,7 @@ protected List<Realm> initRealms() throws Exception {
         Set<String> internalTypes = new HashSet<>();
         List<Realm> realms = new ArrayList<>();
         List<String> kerberosRealmNames = new ArrayList<>();
+        Map<String, Set<String>> nameToRealmIdentifier = new HashMap<>();
         for (RealmConfig.RealmIdentifier identifier: realmsSettings.keySet()) {
             Realm.Factory factory = factories.get(identifier.getType());
             if (factory == null) {
@@ -213,7 +215,10 @@ protected List<Realm> initRealms() throws Exception {
                         "configured");
                 }
             }
-            realms.add(factory.create(config));
+            Realm realm = factory.create(config);
+            nameToRealmIdentifier.computeIfAbsent(realm.name(), k ->
+                new HashSet<>()).add(RealmSettings.realmSettingPrefix(realm.type()) + realm.name());
+            realms.add(realm);
         }
 
         if (!realms.isEmpty()) {
@@ -224,6 +229,13 @@ protected List<Realm> initRealms() throws Exception {
         }
         // 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 + "");
+        }
         return realms;
     }
 

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

@@ -183,8 +183,11 @@ public void init() throws Exception {
         when(licenseState.isAuthAllowed()).thenReturn(true);
         when(licenseState.isApiKeyServiceAllowed()).thenReturn(true);
         when(licenseState.isTokenServiceAllowed()).thenReturn(true);
+        ReservedRealm reservedRealm = mock(ReservedRealm.class);
+        when(reservedRealm.type()).thenReturn("reserved");
+        when(reservedRealm.name()).thenReturn("reserved_realm");
         realms = spy(new TestRealms(Settings.EMPTY, TestEnvironment.newEnvironment(settings), Collections.<String, Realm.Factory>emptyMap(),
-                licenseState, threadContext, mock(ReservedRealm.class), Arrays.asList(firstRealm, secondRealm),
+                licenseState, threadContext, reservedRealm, Arrays.asList(firstRealm, secondRealm),
                 Collections.singletonList(firstRealm)));
 
         auditTrail = mock(AuditTrailService.class);

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

@@ -76,6 +76,7 @@ public void init() throws Exception {
         when(licenseState.isAuthAllowed()).thenReturn(true);
         when(licenseState.allowedRealmType()).thenReturn(AllowedRealmType.ALL);
         when(reservedRealm.type()).thenReturn(ReservedRealm.TYPE);
+        when(reservedRealm.name()).thenReturn("reserved");
     }
 
     public void testWithSettings() throws Exception {
@@ -170,6 +171,20 @@ public void testWithSettingsWithMultipleInternalRealmsOfSameType() throws Except
         }
     }
 
+    public void testWithSettingsWithMultipleRealmsWithSameName() throws Exception {
+        Settings settings = Settings.builder()
+            .put("xpack.security.authc.realms.file.realm_1.order", 0)
+            .put("xpack.security.authc.realms.native.realm_1.order", 1)
+            .put("xpack.security.authc.realms.kerberos.realm_1.order", 2)
+            .put("path.home", createTempDir())
+            .build();
+        Environment env = TestEnvironment.newEnvironment(settings);
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () ->{
+            new Realms(settings, env, factories, licenseState, threadContext, reservedRealm);
+        });
+        assertThat(e.getMessage(), containsString("Found multiple realms configured with the same name"));
+    }
+
     public void testWithEmptySettings() throws Exception {
         Realms realms = new Realms(Settings.EMPTY, TestEnvironment.newEnvironment(Settings.builder().put("path.home",
                 createTempDir()).build()), factories, licenseState, threadContext, reservedRealm);