Watcher add email warning if CSV attachment contains formulas (#44460)

This commit introduces a Warning message to the emails generated by 
Watcher's reporting action. This change complements Kibana's CSV 
formula notifications (see elastic/kibana#37930). 

This is implemented by reading a header (kbn-csv-contains-formulas) 
provided by Kibana to notify to attach the Warning to the email. 
The wording of the warning is borrowed from Kibana's UI and may 
be overridden by a dynamic setting
xpack.notification.reporting.warning.kbn-csv-contains-formulas.text.
This warning is enabled by default, but may be disabled via a 
dynamic setting xpack.notification.reporting.warning.enabled.

x-pack/plugin/watcher/build.gradle

@@ -44,7 +44,7 @@ dependencies {
 
     testCompile 'org.subethamail:subethasmtp:3.1.7'
     // needed for subethasmtp, has @GuardedBy annotation
-    testCompile 'com.google.code.findbugs:jsr305:3.0.1'
+    testCompile 'com.google.code.findbugs:jsr305:3.0.2'
 }
 
 // classes are missing, e.g. com.ibm.icu.lang.UCharacter

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

@@ -284,7 +284,8 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
         Map<String, EmailAttachmentParser> emailAttachmentParsers = new HashMap<>();
         emailAttachmentParsers.put(HttpEmailAttachementParser.TYPE, new HttpEmailAttachementParser(httpClient, templateEngine));
         emailAttachmentParsers.put(DataAttachmentParser.TYPE, new DataAttachmentParser());
-        emailAttachmentParsers.put(ReportingAttachmentParser.TYPE, new ReportingAttachmentParser(settings, httpClient, templateEngine));
+        emailAttachmentParsers.put(ReportingAttachmentParser.TYPE,
+            new ReportingAttachmentParser(settings, httpClient, templateEngine, clusterService.getClusterSettings()));
         EmailAttachmentsParser emailAttachmentsParser = new EmailAttachmentsParser(emailAttachmentParsers);
 
         // conditions
@@ -470,8 +471,7 @@ public List<Setting<?>> getSettings() {
         settings.addAll(HtmlSanitizer.getSettings());
         settings.addAll(JiraService.getSettings());
         settings.addAll(PagerDutyService.getSettings());
-        settings.add(ReportingAttachmentParser.RETRIES_SETTING);
-        settings.add(ReportingAttachmentParser.INTERVAL_SETTING);
+        settings.addAll(ReportingAttachmentParser.getSettings());
 
         // http settings
         settings.addAll(HttpSettings.getSettings());

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

@@ -24,17 +24,26 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Set;
 
 import static javax.mail.Part.ATTACHMENT;
 import static javax.mail.Part.INLINE;
 
 public abstract class Attachment extends BodyPartSource {
 
     private final boolean inline;
+    private final Set<String> warnings;
 
     protected Attachment(String id, String name, String contentType, boolean inline) {
+        this(id, name, contentType, inline, Collections.emptySet());
+    }
+
+    protected Attachment(String id, String name, String contentType, boolean inline, Set<String> warnings) {
         super(id, name, contentType);
         this.inline = inline;
+        assert warnings != null;
+        this.warnings = warnings;
     }
 
     @Override
@@ -53,6 +62,10 @@ public boolean isInline() {
         return inline;
     }
 
+    public Set<String> getWarnings() {
+        return warnings;
+    }
+
     /**
      * intentionally not emitting path as it may come as an information leak
      */
@@ -116,15 +129,15 @@ public static class Bytes extends Attachment {
         private final byte[] bytes;
 
         public Bytes(String id, byte[] bytes, String contentType, boolean inline) {
-            this(id, id, bytes, contentType, inline);
+            this(id, id, bytes, contentType, inline, Collections.emptySet());
         }
 
         public Bytes(String id, String name, byte[] bytes, boolean inline) {
-            this(id, name, bytes, fileTypeMap.getContentType(name), inline);
+            this(id, name, bytes, fileTypeMap.getContentType(name), inline, Collections.emptySet());
         }
 
-        public Bytes(String id, String name, byte[] bytes, String contentType, boolean inline) {
-            super(id, name, contentType, inline);
+        public Bytes(String id, String name, byte[] bytes, String contentType, boolean inline, Set<String> warnings) {
+            super(id, name, contentType, inline, warnings);
             this.bytes = bytes;
         }
 
@@ -213,7 +226,7 @@ protected XContent(String id, ToXContent content, XContentType type) {
         }
 
         protected XContent(String id, String name, ToXContent content, XContentType type) {
-            super(id, name, bytes(name, content, type), mimeType(type), false);
+            super(id, name, bytes(name, content, type), mimeType(type), false, Collections.emptySet());
         }
 
         static String mimeType(XContentType type) {

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.watcher.notification.email;
 
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -16,9 +17,11 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 public class EmailTemplate implements ToXContentObject {
 
@@ -110,19 +113,46 @@ public Email.Builder render(TextTemplateEngine engine, Map<String, Object> model
         if (subject != null) {
             builder.subject(engine.render(subject, model));
         }
-        if (textBody != null) {
-            builder.textBody(engine.render(textBody, model));
-        }
+
+        Set<String> warnings = new HashSet<>(1);
         if (attachments != null) {
             for (Attachment attachment : attachments.values()) {
                 builder.attach(attachment);
+                warnings.addAll(attachment.getWarnings());
             }
         }
+
+        String htmlWarnings = "";
+        String textWarnings = "";
+        if(warnings.isEmpty() == false){
+            StringBuilder textWarningBuilder = new StringBuilder();
+            StringBuilder htmlWarningBuilder = new StringBuilder();
+            warnings.forEach(w ->
+            {
+                if(Strings.isNullOrEmpty(w) == false) {
+                    textWarningBuilder.append(w).append("\n");
+                    htmlWarningBuilder.append(w).append("<br>");
+                }
+            });
+            textWarningBuilder.append("\n");
+            htmlWarningBuilder.append("<br>");
+            htmlWarnings = htmlWarningBuilder.toString();
+            textWarnings = textWarningBuilder.toString();
+        }
+        if (textBody != null) {
+            builder.textBody(textWarnings + engine.render(textBody, model));
+        }
+
         if (htmlBody != null) {
-            String renderedHtml = engine.render(htmlBody, model);
+            String renderedHtml = htmlWarnings + engine.render(htmlBody, model);
             renderedHtml = htmlSanitizer.sanitize(renderedHtml);
             builder.htmlBody(renderedHtml);
         }
+
+        if(htmlBody == null && textBody == null && Strings.isNullOrEmpty(textWarnings) == false){
+            builder.textBody(textWarnings);
+        }
+
         return builder;
     }
 

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

@@ -7,11 +7,13 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.logging.LoggerMessageFormat;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -37,22 +39,39 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class ReportingAttachmentParser implements EmailAttachmentParser<ReportingAttachment> {
 
     public static final String TYPE = "reporting";
 
     // total polling of 10 minutes happens this way by default
-    public static final Setting<TimeValue> INTERVAL_SETTING =
+    static final Setting<TimeValue> INTERVAL_SETTING =
             Setting.timeSetting("xpack.notification.reporting.interval", TimeValue.timeValueSeconds(15), Setting.Property.NodeScope);
-    public static final Setting<Integer> RETRIES_SETTING =
+    static final Setting<Integer> RETRIES_SETTING =
             Setting.intSetting("xpack.notification.reporting.retries", 40, 0, Setting.Property.NodeScope);
 
+    static final Setting<Boolean> REPORT_WARNING_ENABLED_SETTING =
+        Setting.boolSetting("xpack.notification.reporting.warning.enabled", true, Setting.Property.NodeScope, Setting.Property.Dynamic);
+
+    static final Setting.AffixSetting<String> REPORT_WARNING_TEXT =
+        Setting.affixKeySetting("xpack.notification.reporting.warning.", "text",
+            key -> Setting.simpleString(key, Setting.Property.NodeScope, Setting.Property.Dynamic));
+
     private static final ObjectParser<Builder, AuthParseContext> PARSER = new ObjectParser<>("reporting_attachment");
     private static final ObjectParser<KibanaReportingPayload, Void> PAYLOAD_PARSER =
             new ObjectParser<>("reporting_attachment_kibana_payload", true, null);
 
+    static final Map<String, String> WARNINGS = Map.of("kbn-csv-contains-formulas", "Warning: The attachment [%s] contains " +
+        "characters which spreadsheet applications may interpret as formulas. Please ensure that the attachment is safe prior to opening.");
+
     static {
         PARSER.declareInt(Builder::retries, ReportingAttachment.RETRIES);
         PARSER.declareBoolean(Builder::inline, ReportingAttachment.INLINE);
@@ -63,18 +82,52 @@ public class ReportingAttachmentParser implements EmailAttachmentParser<Reportin
         PAYLOAD_PARSER.declareString(KibanaReportingPayload::setPath, new ParseField("path"));
     }
 
+    private static List<Setting<?>> getDynamicSettings() {
+        return Arrays.asList(REPORT_WARNING_ENABLED_SETTING, REPORT_WARNING_TEXT);
+    }
+
+    private static List<Setting<?>> getStaticSettings() {
+        return Arrays.asList(INTERVAL_SETTING, RETRIES_SETTING);
+    }
+
+    public static List<Setting<?>> getSettings() {
+        List<Setting<?>> allSettings = new ArrayList<Setting<?>>(getDynamicSettings());
+        allSettings.addAll(getStaticSettings());
+        return allSettings;
+    }
     private final Logger logger;
     private final TimeValue interval;
     private final int retries;
     private HttpClient httpClient;
     private final TextTemplateEngine templateEngine;
+    private boolean warningEnabled = REPORT_WARNING_ENABLED_SETTING.getDefault(Settings.EMPTY);
+    private final Map<String, String> customWarnings = new ConcurrentHashMap<>(1);
 
-    public ReportingAttachmentParser(Settings settings, HttpClient httpClient, TextTemplateEngine templateEngine) {
+    public ReportingAttachmentParser(Settings settings, HttpClient httpClient, TextTemplateEngine templateEngine,
+                                     ClusterSettings clusterSettings) {
         this.interval = INTERVAL_SETTING.get(settings);
         this.retries = RETRIES_SETTING.get(settings);
         this.httpClient = httpClient;
         this.templateEngine = templateEngine;
         this.logger = LogManager.getLogger(getClass());
+        clusterSettings.addSettingsUpdateConsumer(REPORT_WARNING_ENABLED_SETTING, this::setWarningEnabled);
+        clusterSettings.addAffixUpdateConsumer(REPORT_WARNING_TEXT, this::addWarningText, this::warningValidator);
+    }
+
+    void setWarningEnabled(boolean warningEnabled) {
+        this.warningEnabled = warningEnabled;
+    }
+
+    void addWarningText(String name, String value) {
+        customWarnings.put(name, value);
+    }
+
+    void warningValidator(String name, String value) {
+        if (WARNINGS.keySet().contains(name) == false) {
+            throw new IllegalArgumentException(new ParameterizedMessage(
+                "Warning [{}] is not supported. Only the following warnings are supported [{}]",
+                name, String.join(", ", WARNINGS.keySet())).getFormattedMessage());
+        }
     }
 
     @Override
@@ -139,8 +192,24 @@ public Attachment toAttachment(WatchExecutionContext context, Payload payload, R
                         "method[{}], path[{}], status[{}], body[{}]", context.watch().id(), attachment.id(), request.host(),
                         request.port(), request.method(), request.path(), response.status(), body);
             } else if (response.status() == 200) {
-                return new Attachment.Bytes(attachment.id(), BytesReference.toBytes(response.body()),
-                        response.contentType(), attachment.inline());
+                Set<String> warnings = new HashSet<>(1);
+                if (warningEnabled) {
+                    WARNINGS.forEach((warningKey, defaultWarning) -> {
+                        String[] text = response.header(warningKey);
+                        if (text != null && text.length > 0) {
+                            if (Boolean.valueOf(text[0])) {
+                                String warning = String.format(Locale.ROOT, defaultWarning, attachment.id());
+                                String customWarning = customWarnings.get(warningKey);
+                                if (Strings.isNullOrEmpty(customWarning) == false) {
+                                    warning = String.format(Locale.ROOT, customWarning, attachment.id());
+                                }
+                                warnings.add(warning);
+                            }
+                        }
+                    });
+                }
+                return new Attachment.Bytes(attachment.id(), attachment.id(), BytesReference.toBytes(response.body()),
+                    response.contentType(), attachment.inline(), warnings);
             } else {
                 String body = response.body() != null ? response.body().utf8ToString() : null;
                 String message = LoggerMessageFormat.format("", "Watch[{}] reporting[{}] Unexpected status code host[{}], port[{}], " +