feat/SA-384: cost anomaly parsing and rendering (#41)

* feat: sa-384 cost anomaly event parsed and rendered

Author: wozzer72

README.md

@@ -48,6 +48,10 @@ module "notifications" {
     error_url   = "https://raw.githubusercontent.com/appvia/terraform-aws-notifications/main/resources/posts-attention-icon.png"
     warning_url = "https://raw.githubusercontent.com/appvia/terraform-aws-notifications/main/resources/posts-warning-icon.png"
   }
+
+  # To redirect event URL in post through Identity Center, e.g.:
+  identity_center_start_url = "<your start url>"
+  identity_center_role      = "<your role - consistent across all accounts - namely read only>
 }
 ```
 
@@ -130,7 +134,7 @@ Frequently (quartley at least) check and upgrade:
 | <a name="input_enable_teams"></a> [enable\_teams](#input\_enable\_teams) | To send to teams, set to true | `bool` | `false` | no |
 | <a name="input_identity_center_role"></a> [identity\_center\_role](#input\_identity\_center\_role) | The name of the role to use when redirecting through Identity Center | `string` | `null` | no |
 | <a name="input_identity_center_start_url"></a> [identity\_center\_start\_url](#input\_identity\_center\_start\_url) | The start URL of your Identity Center instance | `string` | `null` | no |
-| <a name="input_post_icons_url"></a> [post\_icons\_url](#input\_post\_icons\_url) | URLs (not base64 encoded!) to publically available icons for highlighting posts of error and/or warning status. Ideally 50px square. Set to non-existent URLs to disable icons | <pre>object({<br>    error_url   = string<br>    warning_url = string<br>  })</pre> | <pre>{<br>  "error_url": "https://sa-251-emblems.s3.eu-west-1.amazonaws.com/attention-50px.png",<br>  "warning_url": "https://sa-251-emblems.s3.eu-west-1.amazonaws.com/warning-50px.png"<br>}</pre> | no |
+| <a name="input_post_icons_url"></a> [post\_icons\_url](#input\_post\_icons\_url) | URLs (not base64 encoded!) to publically available icons for highlighting posts of error and/or warning status. Ideally 50px square. Set to non-existent URLs to disable icons | <pre>object({<br>    error_url   = string<br>    warning_url = string<br>  })</pre> | <pre>{<br>  "error_url": "https://raw.githubusercontent.com/appvia/terraform-aws-notifications/main/resources/posts-attention-icon.png",<br>  "warning_url": "https://raw.githubusercontent.com/appvia/terraform-aws-notifications/main/resources/posts-warning-icon.png"<br>}</pre> | no |
 | <a name="input_slack"></a> [slack](#input\_slack) | The configuration for Slack notifications | <pre>object({<br>    lambda_name = optional(string, "slack-notify")<br>    # The name of the lambda function to create <br>    lambda_description = optional(string, "Lambda function to send slack notifications")<br>    # The description for the slack lambda<br>    secret_name = optional(string)<br>    # An optional secret name in secrets manager to use for the slack configuration <br>    webhook_url = optional(string)<br>    # The webhook url to post to<br>    filter_policy = optional(string)<br>    # An optional SNS subscription filter policy to apply<br>    filter_policy_scope = optional(string)<br>    # If filter policy provided this is the scope of that policy; either "MessageAttributes" (default) or "MessageBody"<br>  })</pre> | `null` | no |
 | <a name="input_sns_topic_policy"></a> [sns\_topic\_policy](#input\_sns\_topic\_policy) | The policy to attach to the sns topic, else we default to account root | `string` | `null` | no |
 | <a name="input_subscribers"></a> [subscribers](#input\_subscribers) | Optional list of custom subscribers to the SNS topic | <pre>map(object({<br>    protocol = string<br>    # The protocol to use. The possible values for this are: sqs, sms, lambda, application. (http or https are partially supported, see below).<br>    endpoint = string<br>    # The endpoint to send data to, the contents will vary with the protocol. (see below for more information)<br>    endpoint_auto_confirms = bool<br>    # Boolean indicating whether the end point is capable of auto confirming subscription e.g., PagerDuty (default is false)<br>    raw_message_delivery = bool<br>    # Boolean indicating whether or not to enable raw message delivery (the original message is directly passed, not wrapped in JSON with the original message in the message property) (default is false)<br>  }))</pre> | `{}` | no |

modules/notify/README.md

@@ -132,7 +132,6 @@ Subsumed by appvia's GNU V3 license; [see license](../../LICENSE).
 |------|-------------|------|---------|:--------:|
 | <a name="input_accounts_id_to_name"></a> [accounts\_id\_to\_name](#input\_accounts\_id\_to\_name) | A mapping of account id and account name - used by notification lamdba to map an account ID to a human readable name | `map(string)` | `{}` | no |
 | <a name="input_architecture"></a> [architecture](#input\_architecture) | Instruction set architecture for your Lambda function. Valid values are "x86\_64" or "arm64". | `string` | `"arm64"` | no |
-| <a name="input_aws_powertools_log_level"></a> [aws\_powertools\_log\_level](#input\_aws\_powertools\_log\_level) | The log level for aws powertools | `string` | `"DEBUG"` | no |
 | <a name="input_aws_powertools_service_name"></a> [aws\_powertools\_service\_name](#input\_aws\_powertools\_service\_name) | The service name to use | `string` | `"appvia-notifications"` | no |
 | <a name="input_cloudwatch_log_group_kms_key_id"></a> [cloudwatch\_log\_group\_kms\_key\_id](#input\_cloudwatch\_log\_group\_kms\_key\_id) | The ARN of the KMS Key to use when encrypting log data for Lambda | `string` | `null` | no |
 | <a name="input_cloudwatch_log_group_retention_in_days"></a> [cloudwatch\_log\_group\_retention\_in\_days](#input\_cloudwatch\_log\_group\_retention\_in\_days) | Specifies the number of days you want to retain log events in log group for Lambda. | `number` | `0` | no |
@@ -143,7 +142,6 @@ Subsumed by appvia's GNU V3 license; [see license](../../LICENSE).
 | <a name="input_enable_slack"></a> [enable\_slack](#input\_enable\_slack) | To send to slack, set to true | `bool` | `false` | no |
 | <a name="input_enable_sns_topic_delivery_status_logs"></a> [enable\_sns\_topic\_delivery\_status\_logs](#input\_enable\_sns\_topic\_delivery\_status\_logs) | Whether to enable SNS topic delivery status logs | `bool` | `false` | no |
 | <a name="input_enable_teams"></a> [enable\_teams](#input\_enable\_teams) | To send to teams, set to true | `bool` | `false` | no |
-| <a name="input_hash_extra"></a> [hash\_extra](#input\_hash\_extra) | The string to add into hashing function. Useful when building same source path for different functions. | `string` | `""` | no |
 | <a name="input_iam_policy_path"></a> [iam\_policy\_path](#input\_iam\_policy\_path) | Path of policies to that should be added to IAM role for Lambda Function | `string` | `null` | no |
 | <a name="input_iam_role_boundary_policy_arn"></a> [iam\_role\_boundary\_policy\_arn](#input\_iam\_role\_boundary\_policy\_arn) | The ARN of the policy that is used to set the permissions boundary for the role | `string` | `null` | no |
 | <a name="input_iam_role_name_prefix"></a> [iam\_role\_name\_prefix](#input\_iam\_role\_name\_prefix) | A unique role name beginning with the specified prefix | `string` | `"lambda"` | no |
@@ -167,7 +165,6 @@ Subsumed by appvia's GNU V3 license; [see license](../../LICENSE).
 | <a name="input_python_runtime"></a> [python\_runtime](#input\_python\_runtime) | The lambda python runtime | `string` | `"python3.12"` | no |
 | <a name="input_recreate_missing_package"></a> [recreate\_missing\_package](#input\_recreate\_missing\_package) | Whether to recreate missing Lambda package if it is missing locally or not | `bool` | `true` | no |
 | <a name="input_reserved_concurrent_executions"></a> [reserved\_concurrent\_executions](#input\_reserved\_concurrent\_executions) | The amount of reserved concurrent executions for this lambda function. A value of 0 disables lambda from being triggered and -1 removes any concurrency limitations | `number` | `-1` | no |
-| <a name="input_slack_emoji"></a> [slack\_emoji](#input\_slack\_emoji) | A custom emoji that will appear on Slack messages | `string` | `":aws:"` | no |
 | <a name="input_sns_topic_feedback_role_description"></a> [sns\_topic\_feedback\_role\_description](#input\_sns\_topic\_feedback\_role\_description) | Description of IAM role to use for SNS topic delivery status logging | `string` | `null` | no |
 | <a name="input_sns_topic_feedback_role_force_detach_policies"></a> [sns\_topic\_feedback\_role\_force\_detach\_policies](#input\_sns\_topic\_feedback\_role\_force\_detach\_policies) | Specifies to force detaching any policies the IAM role has before destroying it. | `bool` | `true` | no |
 | <a name="input_sns_topic_feedback_role_name"></a> [sns\_topic\_feedback\_role\_name](#input\_sns\_topic\_feedback\_role\_name) | Name of the IAM role to use for SNS topic delivery status logging | `string` | `null` | no |

modules/notify/functions/src/msg_parser.py

@@ -38,7 +38,6 @@
 
 class AwsService(Enum):
     """AWS service supported by function"""
-
     cloudwatch = "cloudwatch"
     guardduty = "guardduty"
     securityhub = "securityhub"
@@ -58,10 +57,10 @@ def decrypt_url(encrypted_url: str) -> str:
     except Exception as e:
         raise e
 
-def get_service_url(region: str, service: str, account_id: str) -> str:
+def get_service_url(region: str, service: str) -> str:
     """Get the appropriate service URL for the region
 
-    :param region: name of the AWS region
+    :param region: name of the AWS region - not used on a
     :param service: name of the AWS service
     :returns: AWS console url formatted for the region and service provided
     """
@@ -72,16 +71,26 @@ def get_service_url(region: str, service: str, account_id: str) -> str:
 
         if region.startswith("us-gov-"):
             return f"https://console.amazonaws-us-gov.com/{service_url}"
-        elif IDENTITY_CENTER_URL.__len__ and account_id != None:
-            destination_url = f"https://{region}.console.aws.amazon.com/{service_url}"
-            return f"{IDENTITY_CENTER_URL}/#/console?account_id={account_id}&role_name={IDENTITY_CENTER_ROLE}&destination={urllib.parse.quote(destination_url)}"
         else:
             return f"https://console.aws.amazon.com/{service_url}"
 
     except KeyError:
         print(f"Service {service} is currently not supported")
         raise
 
+def get_target_url(account_id: str, absoluteUrl: str = None) -> str:
+    """Redirect via identity center if defined
+
+    :param account_id: the originating account id to use when using Identity Center redirects
+    :param absoluteUrl: if the service is "absolute", then this is the target url
+    :returns: AWS console url formatted for the given URL, account & role iv via Identity Center
+    """
+    if len(IDENTITY_CENTER_URL) > 0 and account_id != None:
+        return f"{IDENTITY_CENTER_URL}/#/console?account_id={account_id}&role_name={IDENTITY_CENTER_ROLE}&destination={urllib.parse.quote(absoluteUrl)}"
+    else:
+        return f"{absoluteUrl}"
+
+
 class AwsAction(Enum):
     """The individual AWS service types parsed"""
     CLOUDWATCH = "CloudWatch"
@@ -92,6 +101,7 @@ class AwsAction(Enum):
     SAVINGS_PLAN = "SavingsPlan"
     SECURITY_HUB = "SecurityHub"
     DMS = "DMS"
+    COST_ANOMALY = "CostAnomaly"
     UNKNOWN = "Unknown"
 
 class CloudWatchAlarmPriority(Enum):
@@ -124,7 +134,7 @@ def parse_cloudwatch_alarm(message: Dict[str, Any], snsRegion: str) -> Dict[str,
     alarm_arn = message["AlarmArn"]
     alarm_arn_region = message["AlarmArn"].split(":")[3]
 
-    cloudwatch_service_url = get_service_url(region=alarm_arn_region, service="cloudwatch", account_id=account_id)
+    cloudwatch_service_url = get_target_url(account_id=account_id, absoluteUrl=get_service_url(region=alarm_arn_region, service="cloudwatch"))
     cloudwatch_url = f"{cloudwatch_service_url}#alarm:alarmFilter=ANY;name={urllib.parse.quote(name)}"
 
     return {
@@ -187,7 +197,7 @@ def parse_guardduty_finding(message: Dict[str, Any], snsRegion: str) -> Dict[str
     count = service['count']
     guard_duty_id = detail['id']
 
-    guardduty_url = get_service_url(region=region, service="guardduty", account_id=account_id)
+    guardduty_url = get_target_url(account_id=account_id, absoluteUrl=get_service_url(region=region, service="guardduty"))
 
 
     atDT = datetime.fromisoformat(service["eventLastSeen"])
@@ -440,8 +450,8 @@ def parse_security_hub_finding(message: Dict[str, Any], snsRegion: str) -> Dict[
     region = message["FindingId"].split(":")[3]
 
     # not done any real investigztion into the service url yet!!!!!!
-    service_url = get_service_url(region=region, service="securityhub", account_id=account_id)
-    url = f"{service_url}#finding:findindFilter=ANY;rule={urllib.parse.quote(rule_id)}"
+    service_url = get_target_url(account_id=account_id, absoluteUrl=get_service_url(region=region, service="securityhub"))
+    url = f"{service_url}#findings?search=GeneratorId%3D%255Coperator%255C%253AEQUALS%255C%253A{urllib.parse.quote(source)}"
 
     # light touch parse on each resource
     resources:list[Dict[str, Any]] = []
@@ -508,6 +518,87 @@ def parse_dms_notification(message: Dict[str, Any], snsRegion: str) -> Dict[str,
         "at_epoch": floor(atEpoch),
     }
 
+
+class CostAnomalyPriority(Enum):
+    """Maps Cost Anomaly severity state to a normalised 3 level priority"""
+    GOOD = "GOOD"
+    WARNING = "WARNING"
+    ERROR = "ERROR"
+
+def parse_cost_anomaly(message: Dict[str, Any]) -> Dict[str, Any]:
+    """Format Cost Anomaly event into facts
+
+    :params message: SNS message body containing Security Hub event
+    :returns: Cost Anomaly facts
+    """
+
+    # if the anomaly current score is lower than max then it's a warning
+    # if the anomaly current score is equal to max then it's an error (the current event is the max)
+    anomaly_score = message["anomalyScore"]
+    current_anomaly_score = anomaly_score["currentScore"]
+    max_anomaly_score = anomaly_score["maxScore"]
+
+    if current_anomaly_score < max_anomaly_score:
+        priority = CostAnomalyPriority.WARNING.value
+    else:
+        priority = CostAnomalyPriority.ERROR.value
+
+    # to use for the identity center style url
+    originatingAccountId = message["accountId"]
+    originatingUrl = message["anomalyDetailsLink"]
+    url = get_target_url(account_id=originatingAccountId, absoluteUrl=originatingUrl)
+    
+    # start and end
+    startedAt = message["anomalyStartDate"]
+    startedAtDT = datetime.fromisoformat(startedAt)
+    startedAtEpoch = startedAtDT.timestamp()
+    endedAt = message["anomalyEndDate"]
+    endedAtDT = datetime.fromisoformat(endedAt)
+    endedAtEpoch = endedAtDT.timestamp()
+
+    # the anomaly
+    anomaly_id = message["anomalyId"]
+    anomaly_dimension = message["dimensionalValue"]
+    monitor_name = message["monitorName"]
+
+    # the impact
+    impact = message["impact"]
+    expected_spend = impact["totalExpectedSpend"]
+    actual_spend = impact["totalActualSpend"]
+    total_impact = impact["totalImpact"]
+
+    # cost anomaly includes the account id and name that originated the anomaly (linked)
+    # assuming the first root cause
+    rootCauses = message["rootCauses"][0]
+    account_id = rootCauses["linkedAccount"]
+    account_name = rootCauses["linkedAccountName"]
+    region = rootCauses["region"]
+    service = rootCauses["service"]
+    usage = rootCauses["usageType"]
+
+    return {
+        "action": AwsAction.COST_ANOMALY.value,
+        "priority": priority,
+
+        "started": startedAt,
+        "ended": endedAt,
+
+        "anomaly_id": anomaly_id,
+        "monitor_name": monitor_name,
+
+        "expected_spend": expected_spend,
+        "actual_spend": actual_spend,
+        "total_impact": total_impact,
+
+        "account_id": account_id,
+        "account_name": account_name,
+        "region": region,
+        "service": service,
+        "usage": usage,
+
+        "url": url,
+    }
+
 class AwsParsedMessage:
     parsedMsg: Dict[str, Any]
     originalMsg: Dict[str, Any]
@@ -537,7 +628,6 @@ def get_message_payload(
         try:
             message = json.loads(message)
         except json.JSONDecodeError:
-            # logger.debug("SNS Record 'message' is not a structured (JSON) payload; it's just a string message")
             pass
 
     message = cast(Dict[str, Any], message)
@@ -570,6 +660,9 @@ def get_message_payload(
     elif subject.startswith("Savings Plans Coverage Alert:"):
         parsedMsg = parse_aws_savings_plan(subject=subject, message=str(message))
 
+    elif subject.startswith("AWS Cost Management:"):
+        parsedMsg = parse_cost_anomaly(message=message)
+    
     else:
         parsedMsg = {
             "action": AwsAction.UNKNOWN.value,
@@ -612,7 +705,7 @@ def parse_sns(
 
         payload = renderer.payload(
           parsedMessage=parserResults.parsedMsg,
-          originalMessage=message,
+          originalMessage=parserResults.originalMsg,
           subject=subject,
         )
         response = vendor_send_to_function(payload=payload)