Support SNS as a destination

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/AlertingPlugin.kt

@@ -62,6 +62,7 @@ import com.amazon.opendistroforelasticsearch.alerting.resthandler.RestSearchEmai
 import com.amazon.opendistroforelasticsearch.alerting.resthandler.RestSearchEmailGroupAction
 import com.amazon.opendistroforelasticsearch.alerting.resthandler.RestSearchMonitorAction
 import com.amazon.opendistroforelasticsearch.alerting.script.TriggerScript
+import com.amazon.opendistroforelasticsearch.alerting.settings.AWSSettings
 import com.amazon.opendistroforelasticsearch.alerting.settings.AlertingSettings
 import com.amazon.opendistroforelasticsearch.alerting.settings.DestinationSettings
 import com.amazon.opendistroforelasticsearch.alerting.transport.TransportAcknowledgeAlertAction
@@ -254,7 +255,9 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R
                 AlertingSettings.FILTER_BY_BACKEND_ROLES,
                 DestinationSettings.EMAIL_USERNAME,
                 DestinationSettings.EMAIL_PASSWORD,
-                DestinationSettings.ALLOW_LIST
+                DestinationSettings.ALLOW_LIST,
+                AWSSettings.SNS_IAM_USER_ACCESS_KEY,
+                AWSSettings.SNS_IAM_USER_SECRET_KEY
             )
     }
 

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/MonitorRunner.kt

@@ -47,6 +47,7 @@ import com.amazon.opendistroforelasticsearch.alerting.model.action.Action.Compan
 import com.amazon.opendistroforelasticsearch.alerting.model.destination.DestinationContextFactory
 import com.amazon.opendistroforelasticsearch.alerting.script.TriggerExecutionContext
 import com.amazon.opendistroforelasticsearch.alerting.script.TriggerScript
+import com.amazon.opendistroforelasticsearch.alerting.settings.AWSSettings
 import com.amazon.opendistroforelasticsearch.alerting.settings.AlertingSettings.Companion.ALERT_BACKOFF_COUNT
 import com.amazon.opendistroforelasticsearch.alerting.settings.AlertingSettings.Companion.ALERT_BACKOFF_MILLIS
 import com.amazon.opendistroforelasticsearch.alerting.settings.AlertingSettings.Companion.MOVE_ALERTS_BACKOFF_COUNT
@@ -123,6 +124,7 @@ class MonitorRunner(
 
     @Volatile private var destinationSettings = loadDestinationSettings(settings)
     @Volatile private var destinationContextFactory = DestinationContextFactory(client, xContentRegistry, destinationSettings)
+    @Volatile private var awsSettings = AWSSettings.parse(settings)
 
     init {
         clusterService.clusterSettings.addSettingsUpdateConsumer(ALERT_BACKOFF_MILLIS, ALERT_BACKOFF_COUNT) {
@@ -535,6 +537,7 @@ class MonitorRunner(
 
                     val destinationCtx = destinationContextFactory.getDestinationContext(destination)
                     actionOutput[MESSAGE_ID] = destination.publish(
+                        awsSettings,
                         actionOutput[SUBJECT],
                         actionOutput[MESSAGE]!!,
                         destinationCtx

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/model/destination/Destination.kt

@@ -20,13 +20,15 @@ import com.amazon.opendistroforelasticsearch.alerting.destination.message.BaseMe
 import com.amazon.opendistroforelasticsearch.alerting.destination.message.ChimeMessage
 import com.amazon.opendistroforelasticsearch.alerting.destination.message.CustomWebhookMessage
 import com.amazon.opendistroforelasticsearch.alerting.destination.message.EmailMessage
+import com.amazon.opendistroforelasticsearch.alerting.destination.message.SNSMessage
 import com.amazon.opendistroforelasticsearch.alerting.destination.message.SlackMessage
 import com.amazon.opendistroforelasticsearch.alerting.destination.response.DestinationResponse
 import com.amazon.opendistroforelasticsearch.alerting.elasticapi.convertToMap
 import com.amazon.opendistroforelasticsearch.alerting.elasticapi.instant
 import com.amazon.opendistroforelasticsearch.alerting.elasticapi.optionalTimeField
 import com.amazon.opendistroforelasticsearch.alerting.elasticapi.optionalUserField
 import com.amazon.opendistroforelasticsearch.alerting.model.destination.email.Email
+import com.amazon.opendistroforelasticsearch.alerting.settings.AWSSettings
 import com.amazon.opendistroforelasticsearch.alerting.util.DestinationType
 import com.amazon.opendistroforelasticsearch.alerting.util.IndexUtils.Companion.NO_SCHEMA_VERSION
 import com.amazon.opendistroforelasticsearch.commons.authuser.User
@@ -56,6 +58,7 @@ data class Destination(
     val lastUpdateTime: Instant,
     val chime: Chime?,
     val slack: Slack?,
+    val sns: SNS?,
     val customWebhook: CustomWebhook?,
     val email: Email?
 ) : ToXContent {
@@ -96,6 +99,8 @@ data class Destination(
         chime?.writeTo(out)
         out.writeBoolean(slack != null)
         slack?.writeTo(out)
+        out.writeBoolean(sns != null)
+        sns?.writeTo(out)
         out.writeBoolean(customWebhook != null)
         customWebhook?.writeTo(out)
         out.writeBoolean(email != null)
@@ -118,6 +123,7 @@ data class Destination(
         const val LAST_UPDATE_TIME_FIELD = "last_update_time"
         const val CHIME = "chime"
         const val SLACK = "slack"
+        const val SNS_TYPE = "sns"
         const val CUSTOMWEBHOOK = "custom_webhook"
         const val EMAIL = "email"
 
@@ -141,6 +147,7 @@ data class Destination(
             var user: User? = null
             lateinit var type: String
             var slack: Slack? = null
+            var sns: SNS? = null
             var chime: Chime? = null
             var customWebhook: CustomWebhook? = null
             var email: Email? = null
@@ -169,6 +176,9 @@ data class Destination(
                     SLACK -> {
                         slack = Slack.parse(xcp)
                     }
+                    SNS_TYPE -> {
+                        sns = SNS.parse(xcp)
+                    }
                     CUSTOMWEBHOOK -> {
                         customWebhook = CustomWebhook.parse(xcp)
                     }
@@ -197,6 +207,7 @@ data class Destination(
                     lastUpdateTime ?: Instant.now(),
                     chime,
                     slack,
+                    sns,
                     customWebhook,
                     email)
         }
@@ -229,14 +240,15 @@ data class Destination(
                 lastUpdateTime = sin.readInstant(),
                 chime = Chime.readFrom(sin),
                 slack = Slack.readFrom(sin),
+                sns = SNS.readFrom(sin),
                 customWebhook = CustomWebhook.readFrom(sin),
                 email = Email.readFrom(sin)
             )
         }
     }
 
     @Throws(IOException::class)
-    fun publish(compiledSubject: String?, compiledMessage: String, destinationCtx: DestinationContext): String {
+    fun publish(AWSSettings: AWSSettings, compiledSubject: String?, compiledMessage: String, destinationCtx: DestinationContext): String {
         val destinationMessage: BaseMessage
         val responseContent: String
         val responseStatusCode: Int
@@ -255,6 +267,16 @@ data class Destination(
                         .withMessage(messageContent)
                         .build()
             }
+            DestinationType.SNS -> {
+                destinationMessage = SNSMessage.Builder(name)
+                        .withRoleArn(sns?.roleARN)
+                        .withTopicArn(sns?.topicARN)
+                        .withIAMAccessKey(AWSSettings.iamUserAccessKey)
+                        .withIAMSecretKey(AWSSettings.iamUserSecretKey)
+                        .withSubject(compiledSubject)
+                        .withMessage(compiledMessage)
+                        .build()
+            }
             DestinationType.CUSTOM_WEBHOOK -> {
                 destinationMessage = CustomWebhookMessage.Builder(name)
                         .withUrl(customWebhook?.url)
@@ -298,6 +320,7 @@ data class Destination(
         when (type) {
             DestinationType.CHIME -> content = chime?.convertToMap()?.get(type.value)
             DestinationType.SLACK -> content = slack?.convertToMap()?.get(type.value)
+            DestinationType.SNS -> content = sns?.convertToMap()?.get(type.value)
             DestinationType.CUSTOM_WEBHOOK -> content = customWebhook?.convertToMap()?.get(type.value)
             DestinationType.EMAIL -> content = email?.convertToMap()?.get(type.value)
             DestinationType.TEST_ACTION -> content = "dummy"

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/model/destination/SNS.kt

@@ -15,6 +15,10 @@
 
 package com.amazon.opendistroforelasticsearch.alerting.model.destination
 
+import com.amazon.opendistroforelasticsearch.alerting.elasticapi.optionalStringField
+import com.amazon.opendistroforelasticsearch.alerting.util.DestinationType
+import org.elasticsearch.common.io.stream.StreamInput
+import org.elasticsearch.common.io.stream.StreamOutput
 import org.elasticsearch.common.xcontent.ToXContent
 import org.elasticsearch.common.xcontent.XContentBuilder
 import org.elasticsearch.common.xcontent.XContentParser
@@ -23,24 +27,32 @@ import java.io.IOException
 import java.lang.IllegalStateException
 import java.util.regex.Pattern
 
-data class SNS(val topicARN: String, val roleARN: String) : ToXContent {
+data class SNS(val topicARN: String, val roleARN: String?) : ToXContent {
 
     init {
         require(SNS_ARN_REGEX.matcher(topicARN).find()) { "Invalid AWS SNS topic ARN: $topicARN" }
-        require(IAM_ARN_REGEX.matcher(roleARN).find()) { "Invalid AWS role ARN: $roleARN " }
+        if (roleARN != null) {
+            require(IAM_ARN_REGEX.matcher(roleARN).find()) { "Invalid AWS role ARN: $roleARN " }
+        }
     }
 
     override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder {
         return builder.startObject(SNS_TYPE)
                 .field(TOPIC_ARN_FIELD, topicARN)
-                .field(ROLE_ARN_FIELD, roleARN)
+                .optionalStringField(ROLE_ARN_FIELD, roleARN)
                 .endObject()
     }
 
+    @Throws(IOException::class)
+    fun writeTo(out: StreamOutput) {
+        out.writeString(topicARN)
+        out.writeOptionalString(roleARN)
+    }
+
     companion object {
 
         private val SNS_ARN_REGEX = Pattern.compile("^arn:aws(-[^:]+)?:sns:([a-zA-Z0-9-]+):([0-9]{12}):([a-zA-Z0-9-_]+)$")
-        private val IAM_ARN_REGEX = Pattern.compile("^arn:aws(-[^:]+)?:iam::([0-9]{12}):([a-zA-Z0-9-/_]+)$")
+        private val IAM_ARN_REGEX = Pattern.compile("^arn:aws(-[^:]+)?:iam::([0-9]{12}):([a-zA-Z_0-9+=,.@\\-_/]+)$")
 
         const val TOPIC_ARN_FIELD = "topic_arn"
         const val ROLE_ARN_FIELD = "role_arn"
@@ -50,7 +62,7 @@ data class SNS(val topicARN: String, val roleARN: String) : ToXContent {
         @Throws(IOException::class)
         fun parse(xcp: XContentParser): SNS {
             lateinit var topicARN: String
-            lateinit var roleARN: String
+            var roleARN: String? = null
 
             XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
             while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
@@ -64,8 +76,21 @@ data class SNS(val topicARN: String, val roleARN: String) : ToXContent {
                     }
                 }
             }
-            return SNS(requireNotNull(topicARN) { "SNS Action topic_arn is null" },
-                    requireNotNull(roleARN) { "SNS Action role_arn is null" })
+            if (DestinationType.snsUseIamRole) {
+                requireNotNull(roleARN) { "SNS Action role_arn is null" }
+            }
+            return SNS(requireNotNull(topicARN) { "SNS Action topic_arn is null" }, roleARN)
+        }
+
+        @JvmStatic
+        @Throws(IOException::class)
+        fun readFrom(sin: StreamInput): SNS? {
+            return if (sin.readBoolean()) {
+                SNS(
+                    topicARN = sin.readString(),
+                    roleARN = sin.readOptionalString()
+                )
+            } else null
         }
     }
 }

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/settings/AWSSettings.kt

@@ -0,0 +1,51 @@
+/*
+ *   Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file is distributed
+ *   on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *   express or implied. See the License for the specific language governing
+ *   permissions and limitations under the License.
+ */
+
+package com.amazon.opendistroforelasticsearch.alerting.settings
+
+import com.amazon.opendistroforelasticsearch.alerting.util.DestinationType
+import org.elasticsearch.common.settings.SecureSetting
+import org.elasticsearch.common.settings.SecureString
+import org.elasticsearch.common.settings.Settings
+import java.io.IOException
+
+data class AWSSettings(
+    val iamUserAccessKey: SecureString,
+    val iamUserSecretKey: SecureString
+) {
+    companion object {
+        val SNS_IAM_USER_ACCESS_KEY = SecureSetting.secureString(
+                "opendistro.alerting.destination.sns.access.key",
+                null
+        )
+
+        val SNS_IAM_USER_SECRET_KEY = SecureSetting.secureString(
+                "opendistro.alerting.destination.sns.secret.key",
+                null
+        )
+
+        @JvmStatic
+        @Throws(IOException::class)
+        fun parse(settings: Settings): AWSSettings {
+            if (SNS_IAM_USER_ACCESS_KEY.get(settings) == null) {
+                DestinationType.snsUseIamRole = true
+            }
+            return AWSSettings(
+                    SNS_IAM_USER_ACCESS_KEY.get(settings),
+                    SNS_IAM_USER_SECRET_KEY.get(settings)
+            )
+        }
+    }
+}

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/util/DestinationType.kt

@@ -18,11 +18,16 @@ package com.amazon.opendistroforelasticsearch.alerting.util
 enum class DestinationType(val value: String) {
     CHIME("chime"),
     SLACK("slack"),
+    SNS("sns"),
     CUSTOM_WEBHOOK("custom_webhook"),
     EMAIL("email"),
     TEST_ACTION("test_action");
 
     override fun toString(): String {
         return value
     }
+
+    companion object {
+        var snsUseIamRole = false
+    }
 }

alerting/src/main/plugin-metadata/plugin-security.policy

@@ -5,4 +5,28 @@ grant {
 
     permission java.net.SocketPermission "*", "connect,resolve";
     permission java.net.NetPermission "getProxySelector";
+
+    // needed because of problems in ClientConfiguration
+    // TODO: get these fixed in aws sdk
+    permission java.lang.RuntimePermission "accessDeclaredMembers";
+    permission java.lang.RuntimePermission "getClassLoader";
+    permission java.net.SocketPermission "*", "connect";
+    // Needed because of problems in AmazonSNS:
+    // When no region is set on a STSClient instance, the
+    // AWS SDK loads all known partitions from a JSON file and
+    // uses a Jackson's ObjectMapper for that: this one, in
+    // version 2.5.3 with the default binding options, tries
+    // to suppress access checks of ctor/field/method and thus
+    // requires this special permission. AWS must be fixed to
+    // uses Jackson correctly and have the correct modifiers
+    // on binded classes.
+    // TODO: get these fixed in aws sdk
+    // See https://github.com/aws/aws-sdk-java/issues/766
+    permission java.lang.reflect.ReflectPermission "suppressAccessChecks";
+
+    // Below is specific for notification SNS client
+    permission javax.management.MBeanServerPermission "createMBeanServer";
+    permission javax.management.MBeanServerPermission "findMBeanServer";
+    permission javax.management.MBeanPermission "com.amazonaws.metrics.*", "*";
+    permission javax.management.MBeanTrustPermission "register";
 };

alerting/src/test/kotlin/com/amazon/opendistroforelasticsearch/alerting/AlertingRestTestCase.kt

@@ -302,6 +302,7 @@ abstract class AlertingRestTestCase : ODFERestTestCase() {
                 lastUpdateTime = Instant.now(),
                 chime = null,
                 slack = null,
+                sns = null,
                 customWebhook = null,
                 email = null)
     }

alerting/src/test/kotlin/com/amazon/opendistroforelasticsearch/alerting/MonitorRunnerIT.kt

@@ -639,6 +639,7 @@ class MonitorRunnerIT : AlertingRestTestCase() {
                 lastUpdateTime = Instant.now(),
                 chime = null,
                 slack = null,
+                sns = null,
                 customWebhook = null,
                 email = email
         ))

alerting/src/test/kotlin/com/amazon/opendistroforelasticsearch/alerting/action/GetDestinationsResponseTests.kt

@@ -55,6 +55,7 @@ class GetDestinationsResponseTests : ESTestCase() {
                 null,
                 slack,
                 null,
+                null,
                 null)
 
         val req = GetDestinationsResponse(RestStatus.OK, 1, listOf(destination))

alerting/src/test/kotlin/com/amazon/opendistroforelasticsearch/alerting/action/IndexDestinationRequestTests.kt

@@ -49,6 +49,7 @@ class IndexDestinationRequestTests : ESTestCase() {
                         Chime("test.com"),
                         null,
                         null,
+                        null,
                         null
                 )
         )
@@ -88,6 +89,7 @@ class IndexDestinationRequestTests : ESTestCase() {
                         Chime("test.com"),
                         null,
                         null,
+                        null,
                         null
                 )
         )

alerting/src/test/kotlin/com/amazon/opendistroforelasticsearch/alerting/action/IndexDestinationResponseTests.kt

@@ -31,7 +31,7 @@ class IndexDestinationResponseTests : ESTestCase() {
 
         val req = IndexDestinationResponse("1234", 0L, 1L, 2L, RestStatus.CREATED,
                 Destination("1234", 0L, 1, 1, 1, DestinationType.CHIME, "TestChimeDest",
-                        randomUser(), Instant.now(), Chime("test.com"), null, null, null))
+                        randomUser(), Instant.now(), Chime("test.com"), null, null, null, null))
         assertNotNull(req)
 
         val out = BytesStreamOutput()