Improve error message for 6.x style realm settings (#36876)

Realm settings were changed in #30241 in a non-BWC way.
If you try and start a 7.x node using a 6.x config style, then the
default error messages do not adequately describe the cause of
the problem, or the solution.

This change detects the when realms are using the 6.x style and fails
with a specific error message.

This detection is a best-effort, and will detect issues when the
realms have not been modified to use the 7.x style, but may not detect
situations where the configuration was partially changed.

e.g. We can detect this:

    xpack.security.authc:
      realms.pki1.type: pki
      realms.pki1.order: 3
      realms.pki1.ssl.certificate_authorities: [ "ca.crt" ]

But this (where the "order" has been updated, but the "ssl.*" has not)
will fall back to the standard "unknown setting" check

    xpack.security.authc:
      realms.pki.pki1.order: 3
      realms.pki1.ssl.certificate_authorities: [ "ca.crt" ]

Closes: #36026

Author: tvernum

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ssl/SSLService.java

@@ -619,12 +619,19 @@ private static Map<String, Settings> getRealmsSSLSettings(Settings settings) {
         final Map<String, Settings> sslSettings = new HashMap<>();
         final String prefix = "xpack.security.authc.realms.";
         final Map<String, Settings> settingsByRealmType = settings.getGroups(prefix);
-        settingsByRealmType.forEach((realmType, typeSettings) ->
-            typeSettings.getAsGroups().forEach((realmName, realmSettings) -> {
-                Settings realmSSLSettings = realmSettings.getByPrefix("ssl.");
-                // Put this even if empty, so that the name will be mapped to the global SSL configuration
-                sslSettings.put(prefix + realmType + "." + realmName + ".ssl", realmSSLSettings);
-            })
+        settingsByRealmType.forEach((realmType, typeSettings) -> {
+                final Optional<String> nonDottedSetting = typeSettings.keySet().stream().filter(k -> k.indexOf('.') == -1).findAny();
+                if (nonDottedSetting.isPresent()) {
+                    logger.warn("Skipping any SSL configuration from realm [{}{}] because the key [{}] is not in the correct format",
+                        prefix, realmType, nonDottedSetting.get());
+                } else {
+                    typeSettings.getAsGroups().forEach((realmName, realmSettings) -> {
+                        Settings realmSSLSettings = realmSettings.getByPrefix("ssl.");
+                        // Put this even if empty, so that the name will be mapped to the global SSL configuration
+                        sslSettings.put(prefix + realmType + "." + realmName + ".ssl", realmSSLSettings);
+                    });
+                }
+            }
         );
         return sslSettings;
     }

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

@@ -115,6 +115,8 @@
 import org.elasticsearch.xpack.core.security.authc.DefaultAuthenticationFailureHandler;
 import org.elasticsearch.xpack.core.security.authc.InternalRealmsSettings;
 import org.elasticsearch.xpack.core.security.authc.Realm;
+import org.elasticsearch.xpack.core.security.authc.RealmConfig;
+import org.elasticsearch.xpack.core.security.authc.RealmSettings;
 import org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.IndicesAccessControl;
@@ -244,6 +246,7 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
@@ -296,7 +299,7 @@ public Security(Settings settings, final Path configPath) {
         this.env = transportClientMode ? null : new Environment(settings, configPath);
         this.enabled = XPackSettings.SECURITY_ENABLED.get(settings);
         if (enabled && transportClientMode == false) {
-            validateAutoCreateIndex(settings);
+            runStartupChecks(settings);
             // we load them all here otherwise we can't access secure settings since they are closed once the checks are
             // fetched
             final List<BootstrapCheck> checks = new ArrayList<>();
@@ -315,6 +318,12 @@ public Security(Settings settings, final Path configPath) {
             this.bootstrapChecks = Collections.emptyList();
         }
         this.securityExtensions.addAll(extensions);
+
+    }
+
+    private static void runStartupChecks(Settings settings) {
+        validateAutoCreateIndex(settings);
+        validateRealmSettings(settings);
     }
 
     @Override
@@ -781,6 +790,40 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
         return Collections.singletonMap(SetSecurityUserProcessor.TYPE, new SetSecurityUserProcessor.Factory(parameters.threadContext));
     }
 
+    /**
+     * Realm settings were changed in 7.0. This method validates that the settings in use on this node match the new style of setting.
+     * In 6.x a realm config would be
+     * <pre>
+     *   xpack.security.authc.realms.file1.type: file
+     *   xpack.security.authc.realms.file1.order: 0
+     * </pre>
+     * In 7.x this realm should be
+     * <pre>
+     *   xpack.security.authc.realms.file.file1.order: 0
+     * </pre>
+     * If confronted with an old style config, the ES Settings validation would simply fail with an error such as
+     * <em>unknown setting [xpack.security.authc.realms.file1.order]</em>. This validation method provides an error that is easier to
+     * understand and take action on.
+     */
+     static void validateRealmSettings(Settings settings) {
+        final Set<String> badRealmSettings = settings.keySet().stream()
+            .filter(k -> k.startsWith(RealmSettings.PREFIX))
+            .filter(key -> {
+                final String suffix = key.substring(RealmSettings.PREFIX.length());
+                // suffix-part, only contains a single '.'
+                return suffix.indexOf('.') == suffix.lastIndexOf('.');
+            })
+            .collect(Collectors.toSet());
+        if (badRealmSettings.isEmpty() == false) {
+            String sampleRealmSetting = RealmSettings.realmSettingPrefix(new RealmConfig.RealmIdentifier("file", "my_file")) + "order";
+            throw new IllegalArgumentException("Incorrect realm settings found. " +
+                "Realm settings have been changed to include the type as part of the setting key.\n" +
+                "For example '" + sampleRealmSetting + "'\n" +
+                "Found invalid config: " + Strings.collectionToDelimitedString(badRealmSettings, ", ") + "\n" +
+                "Please see the breaking changes documentation."
+            );
+        }
+    }
 
     static boolean indexAuditLoggingEnabled(Settings settings) {
         if (XPackSettings.AUDIT_ENABLED.get(settings)) {