Add SSL/TLS settings for watcher email (#45272)

This change adds a new SSL context

   xpack.notification.email.ssl.*

that supports the standard SSL configuration settings (truststore,
verification_mode, etc). This SSL context is used when configuring
outbound SMTP properties for watcher email notifications.

Resolves: #30307

docs/reference/settings/notification-settings.asciidoc

@@ -76,7 +76,7 @@ corresponding endpoints are whitelisted as well.
 
 [[ssl-notification-settings]]
 :ssl-prefix:             xpack.http
-:component:              {watcher}
+:component:              {watcher} HTTP
 :verifies:
 :server!:
 :ssl-context:            watcher
@@ -215,6 +215,15 @@ HTML feature groups>>.
 Set to `false` to completely disable HTML sanitation. Not recommended.
 Defaults to `true`.
 
+[[ssl-notification-smtp-settings]]
+:ssl-prefix:             xpack.notification.email
+:component:              {watcher} Email
+:verifies:
+:server!:
+:ssl-context:            watcher-email
+
+include::ssl-settings.asciidoc[]
+
 [float]
 [[slack-notification-settings]]
 ==== Slack Notification Settings
@@ -334,4 +343,4 @@ The default event type. Valid values: `trigger`,`resolve`, `acknowledge`.
 `attach_payload`::
 Whether or not to provide the watch payload as context for
 the event by default. Valid values: `true`, `false`.
---
+--

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

@@ -19,6 +19,7 @@ import org.elasticsearch.env.Environment;
 import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.common.socket.SocketAccess;
 import org.elasticsearch.xpack.core.ssl.cert.CertificateInfo;
+import org.elasticsearch.xpack.core.watcher.WatcherField;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.KeyManagerFactory;
@@ -416,6 +417,7 @@ public class SSLService {
         sslSettingsMap.put("xpack.http.ssl", settings.getByPrefix("xpack.http.ssl."));
         sslSettingsMap.putAll(getRealmsSSLSettings(settings));
         sslSettingsMap.putAll(getMonitoringExporterSettings(settings));
+        sslSettingsMap.put(WatcherField.EMAIL_NOTIFICATION_SSL_PREFIX, settings.getByPrefix(WatcherField.EMAIL_NOTIFICATION_SSL_PREFIX));
 
         sslSettingsMap.forEach((key, sslSettings) -> loadConfiguration(key, sslSettings, sslContextHolders));
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/WatcherField.java

@@ -15,5 +15,7 @@ public final class WatcherField {
     public static final Setting<InputStream> ENCRYPTION_KEY_SETTING =
             SecureSetting.secureFile("xpack.watcher.encryption_key", null);
 
+    public static final String EMAIL_NOTIFICATION_SSL_PREFIX = "xpack.notification.email.ssl.";
+
     private WatcherField() {}
 }

x-pack/plugin/watcher/build.gradle

@@ -70,6 +70,10 @@ thirdPartyAudit {
     )
 }
 
+forbiddenPatterns {
+    exclude '**/*.p12'
+}
+
 test {
     /*
      * We have to disable setting the number of available processors as tests in the same JVM randomize processors and will step on each

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/Watcher.java

@@ -265,11 +265,12 @@ public class Watcher extends Plugin implements ActionPlugin, ScriptPlugin, Reloa
 
         new WatcherIndexTemplateRegistry(environment.settings(), clusterService, threadPool, client, xContentRegistry);
 
+        final SSLService sslService = getSslService();
         // http client
-        httpClient = new HttpClient(settings, getSslService(), cryptoService, clusterService);
+        httpClient = new HttpClient(settings, sslService, cryptoService, clusterService);
 
         // notification
-        EmailService emailService = new EmailService(settings, cryptoService, clusterService.getClusterSettings());
+        EmailService emailService = new EmailService(settings, cryptoService, sslService, clusterService.getClusterSettings());
         JiraService jiraService = new JiraService(settings, httpClient, clusterService.getClusterSettings());
         SlackService slackService = new SlackService(settings, httpClient, clusterService.getClusterSettings());
         PagerDutyService pagerDutyService = new PagerDutyService(settings, httpClient, clusterService.getClusterSettings());

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/NotificationService.java

@@ -95,7 +95,7 @@ public abstract class NotificationService<Account> {
         final Settings completeSettings = completeSettingsBuilder.build();
         // obtain account names and create accounts
         final Set<String> accountNames = getAccountNames(completeSettings);
-        this.accounts = createAccounts(completeSettings, accountNames, this::createAccount);
+        this.accounts = createAccounts(completeSettings, accountNames, (name, accountSettings) -> createAccount(name, accountSettings));
         this.defaultAccount = findDefaultAccountOrNull(completeSettings, this.accounts);
     }
 

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/email/Account.java

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.watcher.notification.email;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.SpecialPermission;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.settings.SecureSetting;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
@@ -22,6 +23,8 @@ import javax.mail.Session;
 import javax.mail.Transport;
 import javax.mail.internet.InternetAddress;
 import javax.mail.internet.MimeMessage;
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLSocketFactory;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.security.PrivilegedActionException;
@@ -184,7 +187,7 @@ public class Account {
         final Smtp smtp;
         final EmailDefaults defaults;
 
-        Config(String name, Settings settings) {
+        Config(String name, Settings settings, @Nullable SSLSocketFactory sslSocketFactory) {
             this.name = name;
             profile = Profile.resolve(settings.get("profile"), Profile.STANDARD);
             defaults = new EmailDefaults(name, settings.getAsSettings("email_defaults"));
@@ -193,6 +196,9 @@ public class Account {
                 String msg = "missing required email account setting for account [" + name + "]. 'smtp.host' must be configured";
                 throw new SettingsException(msg);
             }
+            if (sslSocketFactory != null) {
+                smtp.setSocketFactory(sslSocketFactory);
+            }
         }
 
         public Session createSession() {
@@ -220,7 +226,7 @@ public class Account {
             /**
              * Finds a setting, and then a secure setting if the setting is null, or returns null if one does not exist. This differs
              * from other getSetting calls in that it allows for null whereas the other methods throw an exception.
-             *
+             * <p>
              * Note: if your setting was not previously secure, than the string reference that is in the setting object is still
              * insecure. This is only constructing a new SecureString with the char[] of the insecure setting.
              */
@@ -274,6 +280,10 @@ public class Account {
                     settings.put(newKey, TimeValue.parseTimeValue(value, currentKey).millis());
                 }
             }
+
+            public void setSocketFactory(SocketFactory socketFactory) {
+                this.properties.put("mail.smtp.ssl.socketFactory", socketFactory);
+            }
         }
 
         /**

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/email/EmailService.java

@@ -15,15 +15,20 @@ import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.xpack.core.ssl.SSLConfiguration;
+import org.elasticsearch.xpack.core.ssl.SSLConfigurationSettings;
+import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.watcher.crypto.CryptoService;
 import org.elasticsearch.xpack.watcher.notification.NotificationService;
 
 import javax.mail.MessagingException;
-
+import javax.net.ssl.SSLSocketFactory;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+import static org.elasticsearch.xpack.core.watcher.WatcherField.EMAIL_NOTIFICATION_SSL_PREFIX;
+
 /**
  * A component to store email credentials and handle sending email notifications.
  */
@@ -101,13 +106,17 @@ public class EmailService extends NotificationService<Account> {
             Setting.affixKeySetting("xpack.notification.email.account.", "smtp.wait_on_quit",
                     (key) -> Setting.boolSetting(key, true, Property.Dynamic, Property.NodeScope));
 
+    private static final SSLConfigurationSettings SSL_SETTINGS = SSLConfigurationSettings.withPrefix(EMAIL_NOTIFICATION_SSL_PREFIX);
+
     private static final Logger logger = LogManager.getLogger(EmailService.class);
 
     private final CryptoService cryptoService;
+    private final SSLService sslService;
 
-    public EmailService(Settings settings, @Nullable CryptoService cryptoService, ClusterSettings clusterSettings) {
+    public EmailService(Settings settings, @Nullable CryptoService cryptoService, SSLService sslService, ClusterSettings clusterSettings) {
         super("email", settings, clusterSettings, EmailService.getDynamicSettings(), EmailService.getSecureSettings());
         this.cryptoService = cryptoService;
+        this.sslService = sslService;
         // ensure logging of setting changes
         clusterSettings.addSettingsUpdateConsumer(SETTING_DEFAULT_ACCOUNT, (s) -> {});
         clusterSettings.addAffixUpdateConsumer(SETTING_PROFILE, (s, o) -> {}, (s, o) -> {});
@@ -132,10 +141,19 @@ public class EmailService extends NotificationService<Account> {
 
     @Override
     protected Account createAccount(String name, Settings accountSettings) {
-        Account.Config config = new Account.Config(name, accountSettings);
+        Account.Config config = new Account.Config(name, accountSettings, getSmtpSslSocketFactory());
         return new Account(config, cryptoService, logger);
     }
 
+    @Nullable
+    private SSLSocketFactory getSmtpSslSocketFactory() {
+        final SSLConfiguration sslConfiguration = sslService.getSSLConfiguration(EMAIL_NOTIFICATION_SSL_PREFIX);
+        if (sslConfiguration == null) {
+            return null;
+        }
+        return sslService.sslSocketFactory(sslConfiguration);
+    }
+
     public EmailSent send(Email email, Authentication auth, Profile profile, String accountName) throws MessagingException {
         Account account = getAccount(accountName);
         if (account == null) {
@@ -189,6 +207,7 @@ public class EmailService extends NotificationService<Account> {
     public static List<Setting<?>> getSettings() {
         List<Setting<?>> allSettings = new ArrayList<Setting<?>>(EmailService.getDynamicSettings());
         allSettings.addAll(EmailService.getSecureSettings());
+        allSettings.addAll(SSL_SETTINGS.getAllSettings());
         return allSettings;
     }
 

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/actions/email/EmailMessageIdTests.java

@@ -10,6 +10,7 @@ import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.watcher.execution.WatchExecutionContext;
 import org.elasticsearch.xpack.core.watcher.watch.Payload;
 import org.elasticsearch.xpack.watcher.common.text.TextTemplateEngine;
@@ -30,6 +31,7 @@ import java.util.List;
 import java.util.Set;
 
 import static org.hamcrest.Matchers.hasSize;
+import static org.mockito.Mockito.mock;
 
 public class EmailMessageIdTests extends ESTestCase {
 
@@ -56,7 +58,7 @@ public class EmailMessageIdTests extends ESTestCase {
         Set<Setting<?>> registeredSettings = new HashSet<>(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
         registeredSettings.addAll(EmailService.getSettings());
         ClusterSettings clusterSettings = new ClusterSettings(settings, registeredSettings);
-        emailService = new EmailService(settings, null, clusterSettings);
+        emailService = new EmailService(settings, null, mock(SSLService.class), clusterSettings);
         EmailTemplate emailTemplate = EmailTemplate.builder().from("from@example.org").to("to@example.org")
                 .subject("subject").textBody("body").build();
         emailAction = new EmailAction(emailTemplate, null, null, null, null, null);

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/actions/email/EmailSslTests.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.watcher.actions.email;
+
+import org.apache.http.ssl.SSLContextBuilder;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.MockSecureSettings;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.env.TestEnvironment;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ssl.SSLService;
+import org.elasticsearch.xpack.core.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.xpack.core.watcher.watch.Payload;
+import org.elasticsearch.xpack.watcher.common.text.TextTemplateEngine;
+import org.elasticsearch.xpack.watcher.notification.email.EmailService;
+import org.elasticsearch.xpack.watcher.notification.email.EmailTemplate;
+import org.elasticsearch.xpack.watcher.notification.email.HtmlSanitizer;
+import org.elasticsearch.xpack.watcher.notification.email.support.EmailServer;
+import org.elasticsearch.xpack.watcher.test.MockTextTemplateEngine;
+import org.elasticsearch.xpack.watcher.test.WatcherTestUtils;
+import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.Before;
+
+import javax.mail.MessagingException;
+import javax.mail.internet.MimeMessage;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.hamcrest.Matchers.hasSize;
+
+public class EmailSslTests extends ESTestCase {
+
+    private EmailServer server;
+    private TextTemplateEngine textTemplateEngine = new MockTextTemplateEngine();
+    private HtmlSanitizer htmlSanitizer = new HtmlSanitizer(Settings.EMPTY);
+
+    @Before
+    public void startSmtpServer() throws GeneralSecurityException, IOException {
+        final KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
+        final char[] keystorePassword = "test-smtp".toCharArray();
+        try (InputStream is = getDataInputStream("test-smtp.p12")) {
+            keyStore.load(is, keystorePassword);
+        }
+        final SSLContext sslContext = new SSLContextBuilder().loadKeyMaterial(keyStore, keystorePassword).build();
+        server = EmailServer.localhost(logger, sslContext);
+    }
+
+    @After
+    public void stopSmtpServer() {
+        server.stop();
+    }
+
+    public void testFailureSendingMessageToSmtpServerWithUntrustedCertificateAuthority() throws Exception {
+        final Settings.Builder settings = Settings.builder();
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        final ExecutableEmailAction emailAction = buildEmailAction(settings, secureSettings);
+        final WatchExecutionContext ctx = WatcherTestUtils.createWatchExecutionContext();
+        final MessagingException exception = expectThrows(MessagingException.class,
+            () -> emailAction.execute("my_action_id", ctx, Payload.EMPTY));
+        final List<Throwable> allCauses = getAllCauses(exception);
+        assertThat(allCauses, Matchers.hasItem(Matchers.instanceOf(SSLException.class)));
+    }
+
+    public void testCanSendMessageToSmtpServerUsingTrustStore() throws Exception {
+        List<MimeMessage> messages = new ArrayList<>();
+        server.addListener(messages::add);
+        try {
+            final Settings.Builder settings = Settings.builder()
+                .put("xpack.notification.email.ssl.truststore.path", getDataPath("test-smtp.p12"));
+            final MockSecureSettings secureSettings = new MockSecureSettings();
+            secureSettings.setString("xpack.notification.email.ssl.truststore.secure_password", "test-smtp");
+
+            ExecutableEmailAction emailAction = buildEmailAction(settings, secureSettings);
+
+            WatchExecutionContext ctx = WatcherTestUtils.createWatchExecutionContext();
+            emailAction.execute("my_action_id", ctx, Payload.EMPTY);
+
+            assertThat(messages, hasSize(1));
+        } finally {
+            server.clearListeners();
+        }
+    }
+
+    public void testCanSendMessageToSmtpServerByDisablingVerification() throws Exception {
+        List<MimeMessage> messages = new ArrayList<>();
+        server.addListener(messages::add);
+        try {
+            final Settings.Builder settings = Settings.builder().put("xpack.notification.email.ssl.verification_mode", "none");
+            final MockSecureSettings secureSettings = new MockSecureSettings();
+            ExecutableEmailAction emailAction = buildEmailAction(settings, secureSettings);
+
+            WatchExecutionContext ctx = WatcherTestUtils.createWatchExecutionContext();
+            emailAction.execute("my_action_id", ctx, Payload.EMPTY);
+
+            assertThat(messages, hasSize(1));
+        } finally {
+            server.clearListeners();
+        }
+    }
+
+    private ExecutableEmailAction buildEmailAction(Settings.Builder baseSettings, MockSecureSettings secureSettings) {
+        secureSettings.setString("xpack.notification.email.account.test.smtp.secure_password", EmailServer.PASSWORD);
+        Settings settings = baseSettings
+            .put("path.home", createTempDir())
+            .put("xpack.notification.email.account.test.smtp.auth", true)
+            .put("xpack.notification.email.account.test.smtp.user", EmailServer.USERNAME)
+            .put("xpack.notification.email.account.test.smtp.port", server.port())
+            .put("xpack.notification.email.account.test.smtp.host", "localhost")
+            .setSecureSettings(secureSettings)
+            .build();
+
+        Set<Setting<?>> registeredSettings = new HashSet<>(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        registeredSettings.addAll(EmailService.getSettings());
+        ClusterSettings clusterSettings = new ClusterSettings(settings, registeredSettings);
+        SSLService sslService = new SSLService(settings, TestEnvironment.newEnvironment(settings));
+        final EmailService emailService = new EmailService(settings, null, sslService, clusterSettings);
+        EmailTemplate emailTemplate = EmailTemplate.builder().from("from@example.org").to("to@example.org")
+            .subject("subject").textBody("body").build();
+        final EmailAction emailAction = new EmailAction(emailTemplate, null, null, null, null, null);
+        return new ExecutableEmailAction(emailAction, logger, emailService, textTemplateEngine, htmlSanitizer, Collections.emptyMap());
+    }
+
+    private List<Throwable> getAllCauses(Exception exception) {
+        final List<Throwable> allCauses = new ArrayList<>();
+        Throwable cause = exception.getCause();
+        while (cause != null) {
+            allCauses.add(cause);
+            cause = cause.getCause();
+        }
+        return allCauses;
+    }
+
+}
+

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/NotificationServiceTests.java

@@ -13,7 +13,6 @@ import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.watcher.notification.NotificationService;
 
 import java.io.IOException;
 import java.io.InputStream;

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/AccountTests.java

@@ -141,7 +141,7 @@ public class AccountTests extends ESTestCase {
 
         Settings settings = builder.build();
 
-        Account.Config config = new Account.Config(accountName, settings);
+        Account.Config config = new Account.Config(accountName, settings, null);
 
         assertThat(config.profile, is(profile));
         assertThat(config.defaults, equalTo(emailDefaults));
@@ -165,7 +165,7 @@ public class AccountTests extends ESTestCase {
                 .put("smtp.port", server.port())
                 .put("smtp.user", EmailServer.USERNAME)
                 .setSecureSettings(secureSettings)
-                .build()), null, logger);
+                .build(), null), null, logger);
 
         Email email = Email.builder()
                 .id("_id")
@@ -202,7 +202,7 @@ public class AccountTests extends ESTestCase {
                 .put("smtp.port", server.port())
                 .put("smtp.user", EmailServer.USERNAME)
                 .setSecureSettings(secureSettings)
-                .build()), null, logger);
+                .build(), null), null, logger);
 
         Email email = Email.builder()
                 .id("_id")
@@ -240,7 +240,7 @@ public class AccountTests extends ESTestCase {
         Account account = new Account(new Account.Config("default", Settings.builder()
                 .put("smtp.host", "localhost")
                 .put("smtp.port", server.port())
-                .build()), null, logger);
+                .build(), null), null, logger);
 
         Email email = Email.builder()
                 .id("_id")
@@ -264,7 +264,7 @@ public class AccountTests extends ESTestCase {
         Account account = new Account(new Account.Config("default", Settings.builder()
                 .put("smtp.host", "localhost")
                 .put("smtp.port", server.port())
-                .build()), null, logger);
+                .build(), null), null, logger);
 
         Properties mailProperties = account.getConfig().smtp.properties;
         assertThat(mailProperties.get("mail.smtp.connectiontimeout"), is(String.valueOf(TimeValue.timeValueMinutes(2).millis())));
@@ -279,7 +279,7 @@ public class AccountTests extends ESTestCase {
                 .put("smtp.connection_timeout", TimeValue.timeValueMinutes(4))
                 .put("smtp.write_timeout", TimeValue.timeValueMinutes(6))
                 .put("smtp.timeout", TimeValue.timeValueMinutes(8))
-                .build()), null, logger);
+                .build(), null), null, logger);
 
         Properties mailProperties = account.getConfig().smtp.properties;
 
@@ -294,7 +294,7 @@ public class AccountTests extends ESTestCase {
                     .put("smtp.host", "localhost")
                     .put("smtp.port", server.port())
                     .put("smtp.connection_timeout", 4000)
-                    .build()), null, logger);
+                    .build(), null), null, logger);
         });
     }
 

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/AccountsTests.java

@@ -9,6 +9,7 @@ import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ssl.SSLService;
 
 import java.util.HashSet;
 
@@ -16,13 +17,14 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.isOneOf;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.mockito.Mockito.mock;
 
 public class AccountsTests extends ESTestCase {
     public void testSingleAccount() throws Exception {
         Settings.Builder builder = Settings.builder()
                 .put("default_account", "account1");
         addAccountSettings("account1", builder);
-        EmailService service = new EmailService(builder.build(), null,
+        EmailService service = new EmailService(builder.build(), null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         Account account = service.getAccount("account1");
         assertThat(account, notNullValue());
@@ -35,7 +37,7 @@ public class AccountsTests extends ESTestCase {
     public void testSingleAccountNoExplicitDefault() throws Exception {
         Settings.Builder builder = Settings.builder();
         addAccountSettings("account1", builder);
-        EmailService service = new EmailService(builder.build(), null,
+        EmailService service = new EmailService(builder.build(), null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         Account account = service.getAccount("account1");
         assertThat(account, notNullValue());
@@ -51,7 +53,7 @@ public class AccountsTests extends ESTestCase {
         addAccountSettings("account1", builder);
         addAccountSettings("account2", builder);
 
-        EmailService service = new EmailService(builder.build(), null,
+        EmailService service = new EmailService(builder.build(), null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         Account account = service.getAccount("account1");
         assertThat(account, notNullValue());
@@ -70,7 +72,7 @@ public class AccountsTests extends ESTestCase {
         addAccountSettings("account1", builder);
         addAccountSettings("account2", builder);
 
-        EmailService service = new EmailService(builder.build(), null,
+        EmailService service = new EmailService(builder.build(), null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         Account account = service.getAccount("account1");
         assertThat(account, notNullValue());
@@ -88,13 +90,14 @@ public class AccountsTests extends ESTestCase {
         addAccountSettings("account1", builder);
         addAccountSettings("account2", builder);
         ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings()));
-        SettingsException e = expectThrows(SettingsException.class, () -> new EmailService(builder.build(), null, clusterSettings));
+        SettingsException e = expectThrows(SettingsException.class,
+            () -> new EmailService(builder.build(), null, mock(SSLService.class), clusterSettings));
         assertThat(e.getMessage(), is("could not find default account [unknown]"));
     }
 
     public void testNoAccount() throws Exception {
         Settings.Builder builder = Settings.builder();
-        EmailService service = new EmailService(builder.build(), null,
+        EmailService service = new EmailService(builder.build(), null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         expectThrows(IllegalArgumentException.class, () -> service.getAccount(null));
     }
@@ -102,7 +105,8 @@ public class AccountsTests extends ESTestCase {
     public void testNoAccountWithDefaultAccount() throws Exception {
         Settings settings = Settings.builder().put("xpack.notification.email.default_account", "unknown").build();
         ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings()));
-        SettingsException e = expectThrows(SettingsException.class, () -> new EmailService(settings, null, clusterSettings));
+        SettingsException e = expectThrows(SettingsException.class,
+            () -> new EmailService(settings, null, mock(SSLService.class), clusterSettings));
         assertThat(e.getMessage(), is("could not find default account [unknown]"));
     }
 

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/EmailServiceTests.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.watcher.notification.email;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.watcher.common.secret.Secret;
 import org.junit.Before;
 
@@ -32,7 +33,7 @@ public class EmailServiceTests extends ESTestCase {
     public void init() throws Exception {
         account = mock(Account.class);
         service = new EmailService(Settings.builder().put("xpack.notification.email.account.account1.foo", "bar").build(), null,
-                new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings()))) {
+            mock(SSLService.class), new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings()))) {
             @Override
             protected Account createAccount(String name, Settings accountSettings) {
                 return account;
@@ -70,7 +71,7 @@ public class EmailServiceTests extends ESTestCase {
                 .put("xpack.notification.email.account.account5.smtp.wait_on_quit", true)
                 .put("xpack.notification.email.account.account5.smtp.ssl.trust", "host1,host2,host3")
                 .build();
-        EmailService emailService = new EmailService(settings, null,
+        EmailService emailService = new EmailService(settings, null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
 
         Account account1 = emailService.getAccount("account1");

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/ProfileTests.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.watcher.notification.email;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ssl.SSLService;
 
 import javax.mail.BodyPart;
 import javax.mail.Part;
@@ -19,6 +20,7 @@ import java.util.HashSet;
 
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.mockito.Mockito.mock;
 
 public class ProfileTests extends ESTestCase {
 
@@ -40,7 +42,7 @@ public class ProfileTests extends ESTestCase {
                 .put("xpack.notification.email.account.foo.smtp.host", "_host")
                 .build();
 
-        EmailService service = new EmailService(settings, null,
+        EmailService service = new EmailService(settings, null, mock(SSLService.class),
                 new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         Session session = service.getAccount("foo").getConfig().createSession();
         MimeMessage mimeMessage = Profile.STANDARD.toMimeMessage(email, session);
@@ -62,4 +64,4 @@ public class ProfileTests extends ESTestCase {
 
         assertThat("Expected to find an inline attachment in mime message, but didnt", foundInlineAttachment, is(true));
     }
-}
+}

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/notification/email/support/EmailServer.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.watcher.notification.email.support;
 
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.common.Nullable;
 import org.subethamail.smtp.auth.EasyAuthenticationHandlerFactory;
 import org.subethamail.smtp.helper.SimpleMessageListener;
 import org.subethamail.smtp.helper.SimpleMessageListenerAdapter;
@@ -14,8 +15,13 @@ import org.subethamail.smtp.server.SMTPServer;
 import javax.mail.MessagingException;
 import javax.mail.Session;
 import javax.mail.internet.MimeMessage;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
 import java.io.IOException;
 import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.List;
@@ -37,8 +43,8 @@ public class EmailServer {
     private final List<Listener> listeners = new CopyOnWriteArrayList<>();
     private final SMTPServer server;
 
-    public EmailServer(String host, final Logger logger) {
-        server = new SMTPServer(new SimpleMessageListenerAdapter(new SimpleMessageListener() {
+    public EmailServer(String host, @Nullable SSLContext sslContext, final Logger logger) {
+        final SimpleMessageListenerAdapter listener = new SimpleMessageListenerAdapter(new SimpleMessageListener() {
             @Override
             public boolean accept(String from, String recipient) {
                 return true;
@@ -49,9 +55,9 @@ public class EmailServer {
                 try {
                     Session session = Session.getInstance(new Properties());
                     MimeMessage msg = new MimeMessage(session, data);
-                    for (Listener listener : listeners) {
+                    for (Listener listener1 : listeners) {
                         try {
-                            listener.on(msg);
+                            listener1.on(msg);
                         } catch (Exception e) {
                             logger.error("Unexpected failure", e);
                             fail(e.getMessage());
@@ -61,12 +67,33 @@ public class EmailServer {
                     throw new RuntimeException("could not create mime message", me);
                 }
             }
-        }), new EasyAuthenticationHandlerFactory((user, passwd) -> {
+        });
+        final EasyAuthenticationHandlerFactory authentication = new EasyAuthenticationHandlerFactory((user, passwd) -> {
             assertThat(user, is(USERNAME));
             assertThat(passwd, is(PASSWORD));
-        }));
+        });
+        server = new SMTPServer(listener, authentication) {
+            @Override
+            public SSLSocket createSSLSocket(Socket socket) throws IOException {
+                if (sslContext == null) {
+                    return super.createSSLSocket(socket);
+                } else {
+                    SSLSocketFactory factory = sslContext.getSocketFactory();
+                    InetSocketAddress remoteAddress = (InetSocketAddress) socket.getRemoteSocketAddress();
+                    SSLSocket sslSocket = (SSLSocket) factory.createSocket(socket, remoteAddress.getHostString(), socket.getPort(), true);
+                    sslSocket.setUseClientMode(false);
+                    sslSocket.setEnabledCipherSuites(sslSocket.getSupportedCipherSuites());
+                    return sslSocket;
+                }
+            }
+        };
         server.setHostName(host);
         server.setPort(0);
+        if (sslContext != null) {
+            server.setEnableTLS(true);
+            server.setRequireTLS(true);
+            server.setHideTLS(false);
+        }
     }
 
     /**
@@ -93,8 +120,16 @@ public class EmailServer {
         listeners.add(listener);
     }
 
+    public void clearListeners() {
+        this.listeners.clear();
+    }
+
     public static EmailServer localhost(final Logger logger) {
-        EmailServer server = new EmailServer("localhost", logger);
+        return localhost(logger, null);
+    }
+
+    public static EmailServer localhost(final Logger logger, @Nullable SSLContext sslContext) {
+        EmailServer server = new EmailServer("localhost", sslContext, logger);
         server.start();
         return server;
     }

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/test/AbstractWatcherIntegrationTestCase.java

@@ -42,6 +42,7 @@ import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
 import org.elasticsearch.test.store.MockFSIndexStore;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.xpack.core.XPackSettings;
+import org.elasticsearch.xpack.core.ssl.SSLService;
 import org.elasticsearch.xpack.core.watcher.WatcherState;
 import org.elasticsearch.xpack.core.watcher.execution.ExecutionState;
 import org.elasticsearch.xpack.core.watcher.execution.TriggeredWatchStoreField;
@@ -94,6 +95,7 @@ import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.core.Is.is;
 import static org.hamcrest.core.IsNot.not;
+import static org.mockito.Mockito.mock;
 
 @ClusterScope(scope = SUITE, numClientNodes = 0, maxNumDataNodes = 3)
 public abstract class AbstractWatcherIntegrationTestCase extends ESIntegTestCase {
@@ -552,7 +554,8 @@ public abstract class AbstractWatcherIntegrationTestCase extends ESIntegTestCase
     public static class NoopEmailService extends EmailService {
 
         public NoopEmailService() {
-            super(Settings.EMPTY, null, new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
+            super(Settings.EMPTY, null, mock(SSLService.class),
+                new ClusterSettings(Settings.EMPTY, new HashSet<>(EmailService.getSettings())));
         }
 
         @Override