diff --git a/triggers/join-alert/README.md b/triggers/join-alert/README.md new file mode 100644 index 0000000..b0b1494 --- /dev/null +++ b/triggers/join-alert/README.md @@ -0,0 +1,56 @@ +# Tuple Join Alert + +This Tuple trigger sends a desktop notification when a specific person joins a Tuple room. + +## Platform Requirements + +- **macOS**: No additional requirements (uses built-in `osascript`) +- **Windows**: Requires Windows 10+ (uses built-in PowerShell toast notifications) +- **Linux**: Requires `notify-send` (usually part of `libnotify-bin` package) + ```bash + # Ubuntu/Debian + sudo apt-get install libnotify-bin + + # Fedora/RHEL + sudo dnf install libnotify + + # Arch + sudo pacman -S libnotify + ``` + +## Installation + +Set environment variables in your shell profile: + +**macOS/Linux** (`~/.zshrc` or `~/.bashrc`): +```bash +export TUPLE_JOIN_ALERT_NOTIFICATIONS="Smith +Smith:TeamRoom-1 +:TeamRoom-1 +smith@example.com:TeamRoom-1" +``` + +**Windows** (PowerShell profile `$PROFILE`): +```powershell +$env:TUPLE_JOIN_ALERT_NOTIFICATIONS = "Smith +Smith:TeamRoom-1 +:TeamRoom-1 +smith@example.com:TeamRoom-1" +``` + +(Person can be a full name, partial name, or email address. Room is optional - use `person:room` format to specify both) + +## Testing + +Run tests with: +```bash +python3 test_room_joined.py -v +``` + +Tests cover: +- Name matching (exact email, case-insensitive substring) +- Room matching (case-insensitive substring) +- Configuration parsing (`person:room` format) +- Multiple notification combinations +- Cross-platform notification support (macOS, Windows, Linux) +- Edge cases and error handling, including prevention of command injection diff --git a/triggers/join-alert/assets/icon.png b/triggers/join-alert/assets/icon.png new file mode 100644 index 0000000..9640d4d Binary files /dev/null and b/triggers/join-alert/assets/icon.png differ diff --git a/triggers/join-alert/config.json b/triggers/join-alert/config.json new file mode 100644 index 0000000..b2d9bca --- /dev/null +++ b/triggers/join-alert/config.json @@ -0,0 +1,6 @@ +{ + "name": "Join alert", + "description": "Sends a notification when a specific person joins a Tuple room.", + "platforms": ["macos", "linux", "windows"], + "language": "python" +} diff --git a/triggers/join-alert/room-joined b/triggers/join-alert/room-joined new file mode 100755 index 0000000..a4d74bd --- /dev/null +++ b/triggers/join-alert/room-joined @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 + +""" +Tuple Join Alert +See README.md for installation and usage instructions +""" + +import os +import sys +import subprocess +import platform +from typing import List, Tuple, Optional + + +def name_matches(person_filter: str, full_name: str, email: str) -> bool: + return email == person_filter or person_filter.lower() in full_name.lower() + + +def room_matches(room_filter: Optional[str], room_name: str) -> bool: + return not room_filter or room_filter.lower() in room_name.lower() + + +def parse_notification_config(input: str) -> Tuple[str, Optional[str]]: + """ + Parse "person:room" format configuration string. + + Args: + input: Configuration in format "person" or "person:room" + + Returns: + Tuple of (person, room) where room may be None if not specified + """ + if ":" in input: + person, room = input.split(":", 1) + return person, room + else: + return input, None + + +def matches_any_combination( + notifications: List[str], + full_name: str, + email: str, + room_name: str +) -> bool: + """ + Check if current trigger matches any of the configured combinations. + + Args: + notifications: List of notification configurations ("person" or "person:room") + full_name: The trigger's full name + email: The trigger's email + room_name: The trigger's room name + """ + return any( + name_matches(person, full_name, email) and room_matches( + room, room_name) + for combo in notifications + for person, room in [parse_notification_config(combo)] + ) + + +def send_notification(person_name: str, room_name: str) -> None: + """ + Send notification using platform-specific notification system. + + Args: + person_name: Name of the person who joined + room_name: Name of the room + """ + title = "Tuple Join Alert" + message = f"{person_name} joined {room_name}" + system = platform.system() + + try: + if system == "Darwin": # macOS + _send_notification_macos(title, message) + elif system == "Windows": + _send_notification_windows(title, message) + elif system == "Linux": + _send_notification_linux(title, message) + else: + print(f"Warning: Notifications not supported on {system}", file=sys.stderr) + print(f"{title}: {message}") + except Exception as e: + print(f"Error sending notification: {e}", file=sys.stderr) + print(f"{title}: {message}") + + +def _send_notification_macos(title: str, message: str) -> None: + """Send notification on macOS using osascript.""" + def escape_applescript(text: str) -> str: + """Escape double quotes and backslashes to prevent command injection""" + return text.replace('\\', '\\\\').replace('"', '\\"') + + title_escaped = escape_applescript(title) + message_escaped = escape_applescript(message) + + subprocess.run([ + "osascript", + "-e", + f'display notification "{message_escaped}" with title "{title_escaped}" sound name "Hero"' + ], check=True) + + +def _send_notification_windows(title: str, message: str) -> None: + """Send notification on Windows using PowerShell.""" + import html + + # Escape for XML to prevent injection + title_escaped = html.escape(title, quote=True) + message_escaped = html.escape(message, quote=True) + + # Use Windows 10+ toast notifications via PowerShell + ps_script = f''' +[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType = WindowsRuntime] | Out-Null +[Windows.UI.Notifications.ToastNotification, Windows.UI.Notifications, ContentType = WindowsRuntime] | Out-Null +[Windows.Data.Xml.Dom.XmlDocument, Windows.Data.Xml.Dom.XmlDocument, ContentType = WindowsRuntime] | Out-Null + +$APP_ID = 'TupleJoinAlert' + +$template = @" + + + + {title_escaped} + {message_escaped} + + + +"@ + +$xml = New-Object Windows.Data.Xml.Dom.XmlDocument +$xml.LoadXml($template) +$toast = New-Object Windows.UI.Notifications.ToastNotification $xml +[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier($APP_ID).Show($toast) +''' + + subprocess.run( + ["powershell", "-Command", ps_script], + check=True, + capture_output=True + ) + + +def _send_notification_linux(title: str, message: str) -> None: + """ + Send notification on Linux using notify-send. + + Note: Passing arguments as a list (not shell=True) prevents command injection. + """ + subprocess.run( + ["notify-send", "-u", "normal", "-i", "dialog-information", title, message], + check=True + ) + + +def get_array_environment_variable(variable_name: str) -> List[str]: + """ + Get array from environment variable. + + In bash, arrays are passed to child processes as space-separated values + in a single environment variable. This function splits them back into a list. + """ + value = os.environ.get(variable_name, "") + if not value: + return [] + + # Split on newlines or spaces, filter empty strings + # Arrays from bash can be exported with newlines using printf "%s\n" + return [item.strip() for item in value.split("\n") if item.strip()] + + +def main() -> int: + print("Entering script.") + + # Exit if you joined the room yourself + if os.environ.get("TUPLE_TRIGGER_IS_SELF") == "true": + print("Event triggered for self. Exiting.") + return 0 + + # Get TUPLE_TRIGGER environment variables + full_name = os.environ.get("TUPLE_TRIGGER_FULL_NAME", "") + email = os.environ.get("TUPLE_TRIGGER_EMAIL", "") + room_name = os.environ.get("TUPLE_TRIGGER_ROOM_NAME", "") or "Unknown Room" + + # Get notification configurations + notifications = get_array_environment_variable( + "TUPLE_JOIN_ALERT_NOTIFICATIONS") + + if not notifications: + print("Error: TUPLE_JOIN_ALERT_NOTIFICATIONS array must be set", + file=sys.stderr) + return 1 + + if not matches_any_combination(notifications, full_name, email, room_name): + print("Doesn't match any configured combination. Exiting.") + return 0 + + person_name = full_name or email + send_notification(person_name, room_name) + + print("Exiting.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/triggers/join-alert/test_room_joined.py b/triggers/join-alert/test_room_joined.py new file mode 100755 index 0000000..c40b597 --- /dev/null +++ b/triggers/join-alert/test_room_joined.py @@ -0,0 +1,498 @@ +#!/usr/bin/env python3 + +""" +Tests for Tuple Room Notification Trigger +""" + +import unittest +from unittest.mock import patch, MagicMock +import os +import sys +import importlib.util + +# Import the module under test (handling hyphenated filename) +import types +module_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "room-joined") +room_joined = types.ModuleType("room_joined") +with open(module_path, 'r') as f: + exec(f.read(), room_joined.__dict__) + + +class TestNameMatches(unittest.TestCase): + """Test cases for name_matches function""" + + def test_exact_email_match(self): + """Test exact email match""" + self.assertTrue( + room_joined.name_matches( + "john@example.com", + "John Smith", + "john@example.com" + ) + ) + + def test_email_case_sensitive(self): + """Test that email matching is case-sensitive""" + self.assertFalse( + room_joined.name_matches( + "JOHN@example.com", + "John Smith", + "john@example.com" + ) + ) + + def test_full_name_contains_person_case_insensitive(self): + """Test full name contains person (case-insensitive)""" + self.assertTrue( + room_joined.name_matches( + "smith", + "John Smith", + "john@example.com" + ) + ) + + def test_full_name_uppercase(self): + """Test full name matching with uppercase""" + self.assertTrue( + room_joined.name_matches( + "SMITH", + "John Smith", + "john@example.com" + ) + ) + + def test_full_name_mixed_case(self): + """Test full name matching with mixed case""" + self.assertTrue( + room_joined.name_matches( + "SmiTh", + "John Smith", + "john@example.com" + ) + ) + + def test_partial_name_match(self): + """Test partial name matching""" + self.assertTrue( + room_joined.name_matches( + "Joh", + "John Smith", + "john@example.com" + ) + ) + + def test_no_match(self): + """Test no match""" + self.assertFalse( + room_joined.name_matches( + "Jane", + "John Smith", + "john@example.com" + ) + ) + + def test_empty_full_name(self): + """Test with empty full name""" + self.assertFalse( + room_joined.name_matches( + "John", + "", + "john@example.com" + ) + ) + + +class TestRoomMatches(unittest.TestCase): + """Test cases for room_matches function""" + + def test_no_room_filter_matches_all(self): + """Test that None room filter matches all rooms""" + self.assertTrue(room_joined.room_matches(None, "Any Room")) + + def test_empty_room_filter_matches_all(self): + """Test that empty room filter matches all rooms""" + self.assertTrue(room_joined.room_matches("", "Any Room")) + + def test_exact_room_match_case_insensitive(self): + """Test exact room name match (case-insensitive)""" + self.assertTrue( + room_joined.room_matches("project-alpha", "Project-Alpha") + ) + + def test_partial_room_match(self): + """Test partial room name match""" + self.assertTrue( + room_joined.room_matches("alpha", "Project Alpha Team") + ) + + def test_room_match_uppercase(self): + """Test room matching with uppercase filter""" + self.assertTrue( + room_joined.room_matches("ALPHA", "project-alpha") + ) + + def test_room_match_mixed_case(self): + """Test room matching with mixed case""" + self.assertTrue( + room_joined.room_matches("AlPhA", "project-ALPHA-team") + ) + + def test_room_no_match(self): + """Test room no match""" + self.assertFalse( + room_joined.room_matches("beta", "project-alpha") + ) + + +class TestParseNotificationConfig(unittest.TestCase): + """Test cases for parse_notification_config function""" + + def test_person_only(self): + """Test parsing person-only config""" + person, room = room_joined.parse_notification_config("John") + self.assertEqual(person, "John") + self.assertIsNone(room) + + def test_person_and_room(self): + """Test parsing person:room config""" + person, room = room_joined.parse_notification_config("John:alpha") + self.assertEqual(person, "John") + self.assertEqual(room, "alpha") + + def test_email_and_room(self): + """Test parsing email:room config""" + person, room = room_joined.parse_notification_config("john@example.com:HQ-1") + self.assertEqual(person, "john@example.com") + self.assertEqual(room, "HQ-1") + + def test_multiple_colons(self): + """Test parsing config with multiple colons (only first is used as separator)""" + person, room = room_joined.parse_notification_config("john@example.com:room:with:colons") + self.assertEqual(person, "john@example.com") + self.assertEqual(room, "room:with:colons") + + +class TestMatchesAnyCombination(unittest.TestCase): + """Test cases for matches_any_combination function""" + + def test_single_person_match(self): + """Test single person configuration matches""" + self.assertTrue( + room_joined.matches_any_combination( + ["John"], + "John Smith", + "john@example.com", + "Any Room" + ) + ) + + def test_person_and_room_both_match(self): + """Test person:room both match""" + self.assertTrue( + room_joined.matches_any_combination( + ["Smith:alpha"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_person_and_room_person_no_match(self): + """Test person:room where person doesn't match""" + self.assertFalse( + room_joined.matches_any_combination( + ["Jane:alpha"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_person_and_room_room_no_match(self): + """Test person:room where room doesn't match""" + self.assertFalse( + room_joined.matches_any_combination( + ["Smith:beta"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_multiple_configs_first_matches(self): + """Test multiple configs where first one matches""" + self.assertTrue( + room_joined.matches_any_combination( + ["John", "Jane:beta"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_multiple_configs_second_matches(self): + """Test multiple configs where second one matches""" + self.assertTrue( + room_joined.matches_any_combination( + ["Jane", "Smith:alpha"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_multiple_configs_none_match(self): + """Test multiple configs where none match""" + self.assertFalse( + room_joined.matches_any_combination( + ["Jane", "Bob:beta"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_empty_notifications_list(self): + """Test empty notifications list""" + self.assertFalse( + room_joined.matches_any_combination( + [], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + def test_email_exact_match_with_room(self): + """Test exact email match with room filter""" + self.assertTrue( + room_joined.matches_any_combination( + ["john@example.com:alpha"], + "John Smith", + "john@example.com", + "Project Alpha" + ) + ) + + +class TestGetEnvArray(unittest.TestCase): + """Test cases for get_array_environment_variable function""" + + def test_newline_separated(self): + """Test parsing newline-separated array""" + with patch.dict(os.environ, {"TEST_VAR": "value1\nvalue2\nvalue3"}): + result = room_joined.get_array_environment_variable("TEST_VAR") + self.assertEqual(result, ["value1", "value2", "value3"]) + + def test_empty_env_var(self): + """Test empty environment variable""" + with patch.dict(os.environ, {}, clear=True): + result = room_joined.get_array_environment_variable("TEST_VAR") + self.assertEqual(result, []) + + def test_with_extra_whitespace(self): + """Test parsing with extra whitespace""" + with patch.dict(os.environ, {"TEST_VAR": " value1 \n value2 \n value3 "}): + result = room_joined.get_array_environment_variable("TEST_VAR") + self.assertEqual(result, ["value1", "value2", "value3"]) + + def test_with_empty_lines(self): + """Test parsing with empty lines""" + with patch.dict(os.environ, {"TEST_VAR": "value1\n\nvalue2\n\n"}): + result = room_joined.get_array_environment_variable("TEST_VAR") + self.assertEqual(result, ["value1", "value2"]) + + +class TestMain(unittest.TestCase): + """Test cases for main function""" + + def setUp(self): + """Set up test fixtures""" + self.env_patcher = patch.dict(os.environ, { + "TUPLE_TRIGGER_IS_SELF": "false", + "TUPLE_TRIGGER_FULL_NAME": "John Smith", + "TUPLE_TRIGGER_EMAIL": "john@example.com", + "TUPLE_TRIGGER_ROOM_NAME": "Project Alpha", + "TUPLE_JOIN_ALERT_NOTIFICATIONS": "Smith:alpha" + }) + self.env_patcher.start() + + def tearDown(self): + """Tear down test fixtures""" + self.env_patcher.stop() + + def test_main_sends_notification_on_match(self): + """Test main sends notification when match found""" + with patch.object(room_joined, 'send_notification') as mock_send: + result = room_joined.main() + self.assertEqual(result, 0) + mock_send.assert_called_once_with("John Smith", "Project Alpha") + + def test_main_exits_when_self(self): + """Test main exits early when TUPLE_TRIGGER_IS_SELF is true""" + with patch.object(room_joined, 'send_notification') as mock_send: + with patch.dict(os.environ, {"TUPLE_TRIGGER_IS_SELF": "true"}): + result = room_joined.main() + self.assertEqual(result, 0) + mock_send.assert_not_called() + + def test_main_no_notification_when_no_match(self): + """Test main doesn't send notification when no match""" + with patch.object(room_joined, 'send_notification') as mock_send: + with patch.dict(os.environ, {"TUPLE_JOIN_ALERT_NOTIFICATIONS": "Jane:beta"}): + result = room_joined.main() + self.assertEqual(result, 0) + mock_send.assert_not_called() + + def test_main_error_when_no_notifications_config(self): + """Test main returns error when notifications not configured""" + with patch.dict(os.environ, {"TUPLE_JOIN_ALERT_NOTIFICATIONS": ""}, clear=False): + result = room_joined.main() + self.assertEqual(result, 1) + + def test_main_uses_email_when_no_full_name(self): + """Test main uses email for notification when full name is empty""" + with patch.object(room_joined, 'send_notification') as mock_send: + with patch.dict(os.environ, { + "TUPLE_TRIGGER_FULL_NAME": "", + "TUPLE_JOIN_ALERT_NOTIFICATIONS": "john@example.com:alpha" + }): + result = room_joined.main() + self.assertEqual(result, 0) + mock_send.assert_called_once_with("john@example.com", "Project Alpha") + + def test_main_default_room_name(self): + """Test main uses default room name when not provided""" + with patch.object(room_joined, 'send_notification') as mock_send: + with patch.dict(os.environ, {"TUPLE_TRIGGER_ROOM_NAME": ""}, clear=False): + with patch.dict(os.environ, {"TUPLE_JOIN_ALERT_NOTIFICATIONS": "Smith"}): + result = room_joined.main() + self.assertEqual(result, 0) + mock_send.assert_called_once_with("John Smith", "Unknown Room") + + +class TestSendNotification(unittest.TestCase): + """Test cases for send_notification function""" + + @patch('platform.system', return_value='Darwin') + @patch('subprocess.run') + def test_send_notification_macos(self, mock_run, mock_platform): + """Test send_notification calls osascript on macOS""" + room_joined.send_notification("John Smith", "Project Alpha") + + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + self.assertEqual(args[0], "osascript") + self.assertEqual(args[1], "-e") + self.assertIn('John Smith joined Project Alpha', args[2]) + self.assertIn('Tuple Join Alert', args[2]) + + @patch('platform.system', return_value='Windows') + @patch('subprocess.run') + def test_send_notification_windows(self, mock_run, mock_platform): + """Test send_notification calls PowerShell on Windows""" + room_joined.send_notification("John Smith", "Project Alpha") + + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + self.assertEqual(args[0], "powershell") + self.assertEqual(args[1], "-Command") + self.assertIn('John Smith joined Project Alpha', args[2]) + self.assertIn('Tuple Join Alert', args[2]) + + @patch('platform.system', return_value='Linux') + @patch('subprocess.run') + def test_send_notification_linux(self, mock_run, mock_platform): + """Test send_notification calls notify-send on Linux""" + room_joined.send_notification("John Smith", "Project Alpha") + + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + self.assertEqual(args[0], "notify-send") + self.assertIn("Tuple Join Alert", args) + self.assertIn("John Smith joined Project Alpha", args) + + @patch('platform.system', return_value='Darwin') + @patch('subprocess.run') + def test_send_notification_escapes_message(self, mock_run, mock_platform): + """Test send_notification properly escapes special characters on macOS""" + room_joined.send_notification('John "Johnny" Smith', "Alpha's Room") + + # Check that the call was made with properly escaped strings + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + self.assertEqual(args[0], "osascript") + self.assertEqual(args[1], "-e") + # Double quotes should be escaped with backslash + self.assertIn('John \\"Johnny\\" Smith', args[2]) + self.assertIn("Alpha's Room", args[2]) + + @patch('platform.system', return_value='Darwin') + @patch('subprocess.run') + def test_send_notification_escapes_backslashes(self, mock_run, mock_platform): + """Test send_notification properly escapes backslashes to prevent injection""" + room_joined.send_notification('User\\nMalicious', 'Room\\nCode') + + # Check that backslashes are escaped + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + self.assertEqual(args[0], "osascript") + self.assertEqual(args[1], "-e") + # Backslashes should be doubled + self.assertIn('User\\\\nMalicious', args[2]) + self.assertIn('Room\\\\nCode', args[2]) + + @patch('platform.system', return_value='FreeBSD') + @patch('subprocess.run') + def test_send_notification_unsupported_platform(self, mock_run, mock_platform): + """Test send_notification handles unsupported platforms gracefully""" + # Should not raise an exception + room_joined.send_notification("John Smith", "Project Alpha") + + # Should not call subprocess.run + mock_run.assert_not_called() + + @patch('platform.system', return_value='Darwin') + @patch('subprocess.run', side_effect=Exception("Command failed")) + def test_send_notification_error_handling(self, mock_run, mock_platform): + """Test send_notification handles errors gracefully""" + # Should not raise an exception + room_joined.send_notification("John Smith", "Project Alpha") + + @patch('platform.system', return_value='Windows') + @patch('subprocess.run') + def test_send_notification_windows_xml_escaping(self, mock_run, mock_platform): + """Test Windows notification properly escapes XML special characters""" + room_joined.send_notification('Test ', 'Room & "Test"') + + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + self.assertEqual(args[0], "powershell") + ps_script = args[2] + # XML special characters should be escaped + self.assertIn('<script>', ps_script) + self.assertIn('"', ps_script) + self.assertIn('&', ps_script) + # Original unescaped strings should not be in the script + self.assertNotIn('