128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
import logging
|
|
import boto3
|
|
import json
|
|
from typing import Dict, Any
|
|
from http import HTTPStatus
|
|
|
|
logger = logging.getLogger()
|
|
logger.setLevel(logging.INFO)
|
|
|
|
ses_client = boto3.client('ses', region_name='us-east-1')
|
|
FROM_EMAIL = "xmatthewochoa@gmail.com" # Replace with verified SES sender
|
|
TO_EMAIL = ["xmatthewochoa@gmail.com"] # Replace with recipients or fallback list
|
|
|
|
def extract_parameters(parameter_list):
|
|
"""Convert parameter list of dicts into a key-value dictionary."""
|
|
return {param['name']: param.get('value', '') for param in parameter_list}
|
|
|
|
def parse_email_list(value: str) -> list[str]:
|
|
"""Safely parse the email list string."""
|
|
try:
|
|
if not value.strip().startswith("["):
|
|
return TO_EMAIL
|
|
formatted = "[" + ", ".join(f'"{email.strip()}"' for email in value.strip("[]").split(",")) + "]"
|
|
return json.loads(formatted)
|
|
except Exception as e:
|
|
logger.warning("Invalid email list; using default. Error: %s", str(e))
|
|
return TO_EMAIL
|
|
|
|
def send_email(subject: str, body: str, to_email: list[str]) -> None:
|
|
"""Send an email via Amazon SES."""
|
|
try:
|
|
response = ses_client.send_email(
|
|
Source=FROM_EMAIL,
|
|
Destination={'ToAddresses': to_email},
|
|
Message={
|
|
'Subject': {'Data': subject},
|
|
'Body': {
|
|
'Html': {'Data': body},
|
|
'Text': {'Data': 'Personnel check-in summary sent via SES.'}
|
|
}
|
|
}
|
|
)
|
|
logger.info("Email sent successfully: %s", response)
|
|
except Exception as e:
|
|
logger.error("Failed to send email: %s", str(e))
|
|
raise
|
|
|
|
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
|
|
logger.info("Received event: %s", event)
|
|
|
|
try:
|
|
action_group = event['actionGroup']
|
|
function = event['function']
|
|
message_version = event.get('messageVersion', 1)
|
|
|
|
raw_params = event.get('parameters', [])
|
|
parameters = extract_parameters(raw_params)
|
|
|
|
# Parse emails
|
|
email_list = parse_email_list(parameters.get('list_of_emails_address', ''))
|
|
|
|
# Parse JSON Data string
|
|
try:
|
|
data_dict = json.loads(parameters.get('Data', '{}'))
|
|
except json.JSONDecodeError as e:
|
|
logger.warning("Invalid JSON in Data; using empty dict. Error: %s", str(e))
|
|
data_dict = {}
|
|
|
|
email_subject = f'✅ Check-In Notification: {function}'
|
|
param_html_rows = ''.join([
|
|
f"<tr style='border-bottom: 1px solid #eee;'>"
|
|
f"<td style='padding: 12px 10px; font-weight: bold; color: #333;'>{k.replace('_', ' ')}</td>"
|
|
f"<td style='padding: 12px 10px; color: #555;'>{v}</td>"
|
|
f"</tr>"
|
|
for k, v in data_dict.items()
|
|
])
|
|
|
|
email_body = f'''
|
|
<html>
|
|
<body style="font-family: Arial, sans-serif; background-color: #f9f9f9; padding: 20px; margin: 0;">
|
|
<div style="max-width: 700px; margin: auto; background-color: #ffffff; border-radius: 10px; box-shadow: 0 4px 12px rgba(0,0,0,0.05); overflow: hidden;">
|
|
<div style="background-color: #28a745; color: white; padding: 20px; text-align: center;">
|
|
<h2 style="margin: 0;">Personnel Check-In Submission</h2>
|
|
</div>
|
|
<div style="padding: 30px;">
|
|
<p style="font-size: 16px; color: #333;">A personnel check-in has been submitted with the following details:</p>
|
|
<table style="width: 100%; border-collapse: collapse; margin-top: 20px;">
|
|
<tbody>
|
|
{param_html_rows}
|
|
</tbody>
|
|
</table>
|
|
</div>
|
|
<div style="background-color: #f0f0f0; padding: 15px; text-align: center; font-size: 12px; color: #888;">
|
|
© 2025 Security Operations Team. All rights reserved.
|
|
</div>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
'''
|
|
|
|
send_email(email_subject, email_body, email_list)
|
|
|
|
return {
|
|
'response': {
|
|
'actionGroup': action_group,
|
|
'function': function,
|
|
'functionResponse': {
|
|
'responseBody': {
|
|
'TEXT': {'body': f'Check-in email sent to {email_list}.'}
|
|
}
|
|
}
|
|
},
|
|
'messageVersion': message_version
|
|
}
|
|
|
|
except KeyError as e:
|
|
logger.error('Missing required field: %s', str(e))
|
|
return {
|
|
'statusCode': HTTPStatus.BAD_REQUEST,
|
|
'body': f'Error: {str(e)}'
|
|
}
|
|
except Exception as e:
|
|
logger.error('Unexpected error: %s', str(e))
|
|
return {
|
|
'statusCode': HTTPStatus.INTERNAL_SERVER_ERROR,
|
|
'body': 'Internal server error'
|
|
}
|