1  import boto3
   2  import json
   3  import smtplib
   4  import requests
   5  from email.mime.text import MIMEText
   6  
   7  class AlertingHandler:
   8      def __init__(self, alerting_config: dict):
   9          self.alerting_config = alerting_config or {}
  10          self.smtp_credentials = {}
  11  
  12          if "smtp_secret_name" in self.alerting_config:
  13              # Production secure method: Fetch SMTP creds from AWS Secrets Manager
  14              self.smtp_credentials = self.get_smtp_credentials_from_secrets(self.alerting_config["smtp_secret_name"])
  15          elif "smtp_server" in self.alerting_config:
  16              # Local testing: Credentials directly from config
  17              self.smtp_credentials = {
  18                  "smtp_server": self.alerting_config.get("smtp_server"),
  19                  "smtp_port": self.alerting_config.get("smtp_port"),
  20                  "smtp_username": self.alerting_config.get("smtp_username"),
  21                  "smtp_password": self.alerting_config.get("smtp_password"),
  22                  "from_email": self.alerting_config.get("from_email")
  23              }
  24  
  25      def get_smtp_credentials_from_secrets(self, secret_name):
  26          try:
  27              client = boto3.client('secretsmanager')
  28              response = client.get_secret_value(SecretId=secret_name)
  29              secret = json.loads(response['SecretString'])
  30              return {
  31                  "smtp_server": secret["smtp_server"],
  32                  "smtp_port": int(secret["smtp_port"]),
  33                  "smtp_username": secret["smtp_username"],
  34                  "smtp_password": secret["smtp_password"],
  35                  "from_email": secret["from_email"]
  36              }
  37          except Exception as e:
  38              print(f"[ERROR] Failed to fetch SMTP credentials from SecretsManager: {str(e)}")
  39              return {}
  40  
  41      def send_alert(self, subject, message_body):
  42          """Send both Slack and Email alerts if configured."""
  43          if "slack_webhook_url" in self.alerting_config:
  44              try:
  45                  self.send_slack_alert(subject, message_body)
  46              except Exception as e:
  47                  print(f"[WARNING] Failed to send Slack alert: {str(e)}")
  48  
  49          if "email_list" in self.alerting_config:
  50              try:
  51                  self.send_email_alert(subject, message_body)
  52              except Exception as e:
  53                  print(f"[WARNING] Failed to send Email alert: {str(e)}")
  54  
  55      def send_slack_alert(self, subject, message_body):
  56          slack_webhook_url = self.alerting_config.get("slack_webhook_url")
  57          slack_message = {
  58              "text": f"*{subject}*\n{message_body}"
  59          }
  60          headers = {'Content-Type': 'application/json'}
  61  
  62          response = requests.post(slack_webhook_url, headers=headers, data=json.dumps(slack_message))
  63          if response.status_code != 200:
  64              raise Exception(f"Slack webhook failed with status {response.status_code}: {response.text}")
  65  
  66          print(f"[SUCCESS] Slack alert sent.")
  67  
  68      def send_email_alert(self, subject, message_body):
  69          if not self.smtp_credentials:
  70              print("[INFO] SMTP credentials not configured; skipping email alert.")
  71              return
  72  
  73          smtp_server = self.smtp_credentials["smtp_server"]
  74          smtp_port = self.smtp_credentials["smtp_port"]
  75          smtp_username = self.smtp_credentials["smtp_username"]
  76          smtp_password = self.smtp_credentials["smtp_password"]
  77          from_email = self.smtp_credentials["from_email"]
  78          to_emails = self.alerting_config.get("email_list", [])
  79  
  80          msg = MIMEText(message_body)
  81          msg['Subject'] = subject
  82          msg['From'] = from_email
  83          msg['To'] = ", ".join(to_emails)
  84  
  85          with smtplib.SMTP(smtp_server, smtp_port) as server:
  86              server.starttls()
  87              server.login(smtp_username, smtp_password)
  88              server.sendmail(from_email, to_emails, msg.as_string())
  89  
  90          print(f"[SUCCESS] Email alert sent to: {', '.join(to_emails)}")
  91