diff --git a/lambdas/get_location.py b/lambdas/get_location.py new file mode 100644 index 0000000..0320023 --- /dev/null +++ b/lambdas/get_location.py @@ -0,0 +1,88 @@ +import logging +import json +from typing import Dict, Any +from http import HTTPStatus +import urllib.request + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: + """ + AWS Lambda handler for processing Bedrock agent requests and geocoding location names using Nominatim. + + Args: + event (Dict[str, Any]): The Lambda event containing action details + context (Any): The Lambda context object + + Returns: + Dict[str, Any]: Response formatted for Bedrock Agents + """ + try: + action_group = event['actionGroup'] + function = event['function'] + message_version = event.get('messageVersion', 1) + parameters = event.get('parameters', []) + print(parameters) + parameters_dict ={parameter["name"]:parameter["value"] for parameter in parameters} + location_name = parameters_dict.get('location_name') + if not location_name: + raise KeyError("Missing required parameter: 'location_name'") + + # Call Nominatim API to geocode the location + query = urllib.parse.urlencode({ + "q": location_name, + "format": "json", + "limit": 1 + }) + headers = { + "User-Agent": "aws-lambda-geocoder/1.0" + } + url = f"https://nominatim.openstreetmap.org/search?{query}" + + req = urllib.request.Request(url, headers=headers) + with urllib.request.urlopen(req) as response: + response_data = response.read() + print(response_data) + results = json.loads(response_data) + if not results: + raise ValueError(f"No coordinates found for '{location_name}'") + + lat = float(results[0]["lat"]) + lon = float(results[0]["lon"]) + + # Prepare Bedrock-compatible response + response_body = { + 'TEXT': { + 'body': f"Coordinates for '{location_name}': Latitude {lat}, Longitude {lon}" + } + } + + action_response = { + 'actionGroup': action_group, + 'function': function, + 'functionResponse': { + 'responseBody': response_body + } + } + + final_response = { + 'response': action_response, + 'messageVersion': message_version + } + + logger.info('Success: %s', final_response) + return final_response + + 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': f'Internal server error: {str(e)}' + } diff --git a/lambdas/get_time.py b/lambdas/get_time.py new file mode 100644 index 0000000..8846d25 --- /dev/null +++ b/lambdas/get_time.py @@ -0,0 +1,50 @@ +import json +import logging +from datetime import datetime, timedelta, timezone +from typing import Dict, Any +from http import HTTPStatus + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: + try: + action_group = event.get("actionGroup") + function = event.get("function") + message_version = event.get("messageVersion", 1) + + # Get current UTC time + utc_now = datetime.now(timezone.utc) + + # Assume CET is UTC+2 for summer (CEST) + # Use UTC+1 if you want standard time + cet_offset = timezone(timedelta(hours=2)) # Change to +1 in winter + cet_now = utc_now.astimezone(cet_offset) + + response_body = { + "TEXT": { + "body": ( + f"Current time:\n" + f"- UTC: {utc_now.strftime('%Y-%m-%d %H:%M:%S %Z')}\n" + f"- CET (UTC+2): {cet_now.strftime('%Y-%m-%d %H:%M:%S %Z')}" + ) + } + } + + return { + "response": { + "actionGroup": action_group, + "function": function, + "functionResponse": { + "responseBody": response_body + } + }, + "messageVersion": message_version + } + + except Exception as e: + logger.error("Unexpected error: %s", str(e)) + return { + "statusCode": HTTPStatus.INTERNAL_SERVER_ERROR, + "body": f"Internal server error: {str(e)}" + } diff --git a/lambdas/save_check_in_details.py b/lambdas/save_check_in_details.py new file mode 100644 index 0000000..6933e61 --- /dev/null +++ b/lambdas/save_check_in_details.py @@ -0,0 +1,79 @@ +import json +import logging +import boto3 +from typing import Dict, Any +from http import HTTPStatus + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +dynamodb = boto3.resource('dynamodb') +table = dynamodb.Table('checkin_table') + +def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: + try: + action_group = event.get('actionGroup') + function = event.get('function') + message_version = event.get('messageVersion', 1) + parameters = event.get('parameters', []) + + # Extract 'data' parameter (stringified JSON) + data_param = next((p for p in parameters if p["name"] == "data"), None) + if not data_param: + raise KeyError("Missing 'data' parameter.") + + data = json.loads(data_param["value"]) + + required_fields = [ + "CheckInDate", "UserID", "CheckInTime", "LocationGps", + "LocationLink", "LocationName", "Organization", "Position" + ] + missing = [f for f in required_fields if f not in data] + if missing: + raise KeyError(f"Missing fields in data: {missing}") + + # Prepare item for DynamoDB + item = { + "CheckInDate": data["CheckInDate"], + "UserID": data["UserID"], + "Position": data["Position"], + "Organization": data["Organization"], + "CheckInTime": data["CheckInTime"], + "LocationGps": data["LocationGps"], + "LocationName": data["LocationName"], + "LocationLink": data["LocationLink"], + "Notes": data.get("Notes", "") + } + + table.put_item(Item=item) + + response_body = { + "TEXT": { + "body": f"Check-in recorded successfully for {item['UserID']}." + } + } + action_response = { + "actionGroup": action_group, + "function": function, + "functionResponse": { + "responseBody": response_body + } + } + return { + "response": action_response, + "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": f"Internal server error: {str(e)}" + } diff --git a/lambdas/send_email.py b/lambdas/send_email.py new file mode 100644 index 0000000..277dd48 --- /dev/null +++ b/lambdas/send_email.py @@ -0,0 +1,127 @@ +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"" + f"{k.replace('_', ' ')}" + f"{v}" + f"" + for k, v in data_dict.items() + ]) + + email_body = f''' + + +
+
+

Personnel Check-In Submission

+
+
+

A personnel check-in has been submitted with the following details:

+ + + {param_html_rows} + +
+
+
+ © 2025 Security Operations Team. All rights reserved. +
+
+ + + ''' + + 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' + }