71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
import logging
|
|
|
|
def validate_slack_event(event_data, max_message_length, valid_event_types=None):
|
|
"""
|
|
Validate incoming Slack event to ensure it's a legitimate message.
|
|
|
|
Args:
|
|
event_data (dict): The incoming event data from Slack
|
|
max_message_length (int): Maximum allowed length for messages
|
|
valid_event_types (list, optional): List of valid event types. Defaults to ['message', 'app_mention', 'app_home_opened', 'event_callback']
|
|
|
|
Returns:
|
|
bool: True if the event is valid, False otherwise
|
|
"""
|
|
# Set default valid_event_types if none provided
|
|
if valid_event_types is None:
|
|
valid_event_types = ['message', 'app_mention', 'app_home_opened', 'event_callback']
|
|
|
|
# Check if event_data is a dictionary
|
|
if not isinstance(event_data, dict):
|
|
logging.warning(f"Invalid event: Not a dictionary. Received type: {type(event_data)}")
|
|
return False
|
|
|
|
# Check for 'event' key
|
|
if 'event' not in event_data:
|
|
logging.warning("Invalid event: Missing 'event' key")
|
|
return False
|
|
|
|
# Check that event is a dictionary
|
|
if not isinstance(event_data['event'], dict):
|
|
logging.warning(f"Invalid event: 'event' is not a dictionary. Received type: {type(event_data['event'])}")
|
|
return False
|
|
|
|
# Check for event ID
|
|
if 'event_id' not in event_data:
|
|
logging.warning("Invalid event: Missing 'event_id'")
|
|
return False
|
|
|
|
# Validate event type
|
|
event_type = event_data.get('type')
|
|
if event_type not in valid_event_types:
|
|
logging.warning(f"Invalid event type: {event_type}")
|
|
return False
|
|
|
|
# Basic message validation
|
|
message = event_data['event']
|
|
|
|
# Ensure message has required keys
|
|
required_keys = ['channel', 'user', 'text', 'ts']
|
|
for key in required_keys:
|
|
if key not in message:
|
|
logging.warning(f"Invalid message: Missing required key '{key}'")
|
|
return False
|
|
|
|
# Validate channel and user IDs (basic length and format check)
|
|
if not (isinstance(message['channel'], str) and len(message['channel']) > 0):
|
|
logging.warning("Invalid channel ID")
|
|
return False
|
|
|
|
if not (isinstance(message['user'], str) and len(message['user']) > 0):
|
|
logging.warning("Invalid user ID")
|
|
return False
|
|
|
|
# Check message length to prevent extremely large messages
|
|
if len(message.get('text', '')) > max_message_length:
|
|
logging.warning(f"Message exceeds maximum length of {max_message_length} characters")
|
|
return False
|
|
|
|
# Additional security checks can be added here
|
|
|
|
return True |