diff --git a/keep/api/bl/enrichments_bl.py b/keep/api/bl/enrichments_bl.py index edb6e53449..fc2683140f 100644 --- a/keep/api/bl/enrichments_bl.py +++ b/keep/api/bl/enrichments_bl.py @@ -6,7 +6,7 @@ import uuid from uuid import UUID -import celpy +import CelPy as celpy import chevron import json5 from elasticsearch import NotFoundError diff --git a/keep/providers/base/base_provider.py b/keep/providers/base/base_provider.py index 47c701156d..eac4b57047 100644 --- a/keep/providers/base/base_provider.py +++ b/keep/providers/base/base_provider.py @@ -187,6 +187,8 @@ def notify(self, **kwargs): Args: **kwargs (dict): The provider context (with statement) """ + # Clear results to avoid accumulation when provider is reused across actions/steps + self.results = [] # Pop Keep-internal fields before passing kwargs to the provider enrich_alert = kwargs.pop("enrich_alert", []) enrich_incident = kwargs.pop("enrich_incident", []) @@ -387,6 +389,8 @@ def _query(self, **kwargs: dict): raise NotImplementedError("query() method not implemented") def query(self, **kwargs: dict): + # Clear results to avoid accumulation when provider is reused across actions/steps + self.results = [] # Pop Keep-internal fields before passing kwargs to the provider enrich_alert = kwargs.pop("enrich_alert", []) audit_enabled = bool(kwargs.pop("audit_enabled", True)) diff --git a/keep/providers/cisco_webex_provider/__init__.py b/keep/providers/cisco_webex_provider/__init__.py new file mode 100644 index 0000000000..cb40797e77 --- /dev/null +++ b/keep/providers/cisco_webex_provider/__init__.py @@ -0,0 +1,3 @@ +from .cisco_webex_provider import CiscoWebexProvider + +__all__ = ["CiscoWebexProvider"] diff --git a/keep/providers/cisco_webex_provider/cisco_webex_provider.py b/keep/providers/cisco_webex_provider/cisco_webex_provider.py new file mode 100644 index 0000000000..b6d30ded9a --- /dev/null +++ b/keep/providers/cisco_webex_provider/cisco_webex_provider.py @@ -0,0 +1,178 @@ +""" +CiscoWebexProvider is a notification provider for Cisco Webex. +API: https://developer.webex.com/docs/api/v1/messages +""" + +import dataclasses +from typing import Optional + +import pydantic +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig + + +@pydantic.dataclasses.dataclass +class CiscoWebexProviderAuthConfig: + """Cisco Webex authentication configuration.""" + + access_token: str = dataclasses.field( + metadata={ + "required": True, + "description": "Cisco Webex Bot Access Token (Bearer token)", + "sensitive": True, + } + ) + room_id: str = dataclasses.field( + metadata={ + "required": True, + "description": "Cisco Webex Room ID (e.g., YJIxqqqBRfwL9k3n8z3k9z3k9z3k9z3k)", + } + ) + + +class CiscoWebexProvider(BaseProvider): + """Send alert messages to Cisco Webex rooms via bot API.""" + + PROVIDER_DISPLAY_NAME = "Cisco Webex" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://webex.com/" + PROVIDER_DESCRIPTION = "Send Keep alerts to Cisco Webex rooms via bot API" + IS_TESTABLE = True + WEBEX_API = "https://webexapis.com/v1/messages" + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = CiscoWebexProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No cleanup needed.""" + pass + + def _notify( + self, + message: str = "", + title: str = "", + **kwargs: dict, + ): + """Send notification to Cisco Webex room.""" + self.logger.debug("Sending message to Cisco Webex") + + if not message: + raise ProviderException( + f"{self.__class__.__name__} failed to send message: message body is required" + ) + + display_message = message + if title: + display_message = f"## {title}\n\n{message}" + + url = CiscoWebexProvider.WEBEX_API + payload = { + "roomId": self.authentication_config.room_id, + "text": display_message, + "markdown": display_message, + } + + headers = { + "Authorization": f"Bearer {self.authentication_config.access_token}", + "Content-Type": "application/json", + } + + try: + response = requests.post( + url, + json=payload, + headers=headers, + timeout=30, + ) + + if response.status_code not in (200, 201): + error_msg = response.text + raise ProviderException( + f"{self.__class__.__name__} failed to send Webex message: " + f"HTTP {response.status_code} - {error_msg[:200]}" + ) + + self.logger.info(f"Message sent to Webex room {self.authentication_config.room_id}") + return True + + except requests.exceptions.ConnectionError as e: + raise ProviderException( + f"{self.__class__.__name__} failed to connect to Webex: {str(e)}" + ) + except requests.exceptions.Timeout: + raise ProviderException( + f"{self.__class__.__name__} connection to Webex timed out" + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send Webex message: {str(e)}" + ) + + def test(self): + """Test the Cisco Webex connection.""" + self.logger.debug("Testing Cisco Webex connection") + + if not self.authentication_config: + self.validate_config() + + url = CiscoWebexProvider.WEBEX_API + payload = { + "roomId": self.authentication_config.room_id, + "text": "KeepHQ test notification - connection successful!", + } + + headers = { + "Authorization": f"Bearer {self.authentication_config.access_token}", + "Content-Type": "application/json", + } + + try: + response = requests.post( + url, + json=payload, + headers=headers, + timeout=30, + ) + + if response.status_code in (200, 201): + return { + "ok": True, + "message": "Webex connection successful - test message sent", + } + elif response.status_code == 401: + return { + "ok": False, + "message": "Webex connection test failed: Invalid token (401 Unauthorized)", + } + elif response.status_code == 404: + return { + "ok": False, + "message": "Webex connection test failed: Room not found", + } + else: + return { + "ok": False, + "message": f"Webex connection test failed: HTTP {response.status_code} - {response.text[:200]}", + } + + except requests.exceptions.ConnectionError: + return { + "ok": False, + "message": f"Cannot connect to Webex API", + } + except Exception as e: + return { + "ok": False, + "message": f"Webex connection test failed: {str(e)}", + } diff --git a/keep/providers/flock_provider/__init__.py b/keep/providers/flock_provider/__init__.py new file mode 100644 index 0000000000..889146309c --- /dev/null +++ b/keep/providers/flock_provider/__init__.py @@ -0,0 +1 @@ +from .flock_provider import FlockProvider diff --git a/keep/providers/flock_provider/assets/flock.svg b/keep/providers/flock_provider/assets/flock.svg new file mode 100644 index 0000000000..80ece107aa --- /dev/null +++ b/keep/providers/flock_provider/assets/flock.svg @@ -0,0 +1,4 @@ + + + F + diff --git a/keep/providers/flock_provider/flock_provider.py b/keep/providers/flock_provider/flock_provider.py new file mode 100644 index 0000000000..d3ec321fe0 --- /dev/null +++ b/keep/providers/flock_provider/flock_provider.py @@ -0,0 +1,189 @@ +""" +FlockProvider is a class that provides incoming webhook notification for Flock. +Flock is a team collaboration platform. + +API: https://dev.flock.com/webhooks +""" + +import dataclasses +from typing import Optional + +import pydantic +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig + + +@pydantic.dataclasses.dataclass +class FlockProviderAuthConfig: + """Flock authentication configuration.""" + + webhook_url: str = dataclasses.field( + metadata={ + "required": True, + "description": "Flock Webhook URL (e.g., https://api.flock.com/chat.send?token=YOUR_TOKEN)", + } + ) + + channel: str = dataclasses.field( + metadata={ + "required": True, + "description": "Flock channel ID or user ID to send messages to", + "hint": "Find channel ID in channel settings", + } + ) + + +class FlockProvider(BaseProvider): + """Send alert messages to Flock channels via incoming webhook.""" + + PROVIDER_DISPLAY_NAME = "Flock" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://flock.com/" + PROVIDER_DESCRIPTION = "Send Keep alerts to Flock channels via incoming webhooks" + IS_TESTABLE = True + FLOCK_API = "https://api.flock.com" + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = FlockProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No cleanup needed.""" + pass + + def _parse_webhook_url(self): + """Parse the webhook URL to extract base URL and token.""" + url = self.authentication_config.webhook_url + # Format: https://api.flock.com/chat.send?token=TOKEN + if "?token=" in url: + base_url = url.split("?")[0] + token = url.split("token=")[1].split("&")[0] + else: + base_url = "https://api.flock.com/chat.send" + token = "" + return base_url, token + + def _notify( + self, + message: str = "", + **kwargs: dict, + ): + """ + Send notification message to Flock channel. + + Args: + message (str): Message body to send + """ + self.logger.debug("Sending message to Flock channel") + + if not message: + raise ProviderException( + f"{self.__class__.__name__} failed to send message: message body is required" + ) + + base_url, token = self._parse_webhook_url() + + payload = { + "token": token, + "channel": self.authentication_config.channel, + "text": message, + "type": "text", + } + + try: + response = requests.post( + base_url, + json=payload, + timeout=30, + ) + + if response.status_code != 200: + error_msg = response.text + raise ProviderException( + f"{self.__class__.__name__} failed to send Flock message: " + f"HTTP {response.status_code} - {error_msg[:200]}" + ) + + result = response.json() + if result.get("status") != 1: + raise ProviderException( + f"{self.__class__.__name__} failed to send Flock message: {result.get('error', 'Unknown error')}" + ) + + self.logger.info(f"Message sent to Flock channel {self.authentication_config.channel}") + return True + + except requests.exceptions.ConnectionError as e: + raise ProviderException( + f"{self.__class__.__name__} failed to connect to Flock: {str(e)}" + ) + except requests.exceptions.Timeout: + raise ProviderException( + f"{self.__class__.__name__} connection to Flock timed out" + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send Flock message: {str(e)}" + ) + + def test(self): + """Test the Flock connection by sending a test message.""" + self.logger.debug("Testing Flock connection") + + if not self.authentication_config: + self.validate_config() + + base_url, token = self._parse_webhook_url() + + payload = { + "token": token, + "channel": self.authentication_config.channel, + "text": "KeepHQ test message - connection successful!", + "type": "text", + } + + try: + response = requests.post( + base_url, + json=payload, + timeout=30, + ) + + if response.status_code == 200: + result = response.json() + if result.get("status") == 1: + return { + "ok": True, + "message": "Flock connection successful - test message sent", + } + else: + return { + "ok": False, + "message": f"Flock test failed: {result.get('error', 'Unknown error')}", + } + else: + return { + "ok": False, + "message": f"Flock connection test failed: HTTP {response.status_code} - {response.text[:200]}", + } + + except requests.exceptions.ConnectionError: + return { + "ok": False, + "message": "Cannot connect to Flock", + } + except Exception as e: + return { + "ok": False, + "message": f"Flock connection test failed: {str(e)}", + } diff --git a/keep/providers/gotify_provider/__init__.py b/keep/providers/gotify_provider/__init__.py new file mode 100644 index 0000000000..873aeb78a7 --- /dev/null +++ b/keep/providers/gotify_provider/__init__.py @@ -0,0 +1 @@ +from .gotify_provider import GotifyProvider diff --git a/keep/providers/gotify_provider/assets/gotify.svg b/keep/providers/gotify_provider/assets/gotify.svg new file mode 100644 index 0000000000..f093e58b15 --- /dev/null +++ b/keep/providers/gotify_provider/assets/gotify.svg @@ -0,0 +1,4 @@ + + + + diff --git a/keep/providers/gotify_provider/gotify_provider.py b/keep/providers/gotify_provider/gotify_provider.py new file mode 100644 index 0000000000..6a8eacdca9 --- /dev/null +++ b/keep/providers/gotify_provider/gotify_provider.py @@ -0,0 +1,173 @@ +""" +GotifyProvider is an interface for sending push notifications via Gotify. +Gotify is a self-hosted push notification server. + +API: https://gotify.net/docs/pushmsg +""" + +import dataclasses +from typing import Optional + +import pydantic +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig + + +@pydantic.dataclasses.dataclass +class GotifyProviderAuthConfig: + """Gotify authentication configuration.""" + + server_url: str = dataclasses.field( + metadata={ + "required": True, + "description": "Gotify server URL (e.g., https://gotify.example.com)", + } + ) + + token: str = dataclasses.field( + metadata={ + "required": True, + "description": "Gotify application token", + "sensitive": True, + } + ) + + +class GotifyProvider(BaseProvider): + """Send push notifications via Gotify.""" + + PROVIDER_DISPLAY_NAME = "Gotify" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://gotify.net/" + PROVIDER_DESCRIPTION = "Send Keep alerts via Gotify push notifications" + IS_TESTABLE = True + + GOTIFY_API = "https://gotify.net/docs/pushmsg" + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = GotifyProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No cleanup needed.""" + pass + + def _notify( + self, + message: str = "", + title: str = "", + priority: int = 0, + **kwargs: dict, + ): + """ + Send notification via Gotify. + + Args: + message (str): Message body to send + title (str): Message title (defaults to alert name) + priority (int): Message priority (0-10), defaults to 0 + """ + self.logger.debug("Sending message to Gotify") + + if not message: + raise ProviderException( + f"{self.__class__.__name__} failed to send message: message body is required" + ) + + url = f"{self.authentication_config.server_url}/message?token={self.authentication_config.token}" + + body = { + "message": message, + "priority": priority, + } + if title: + body["title"] = title + + try: + response = requests.post( + url, + json=body, + timeout=30, + ) + + if response.status_code not in (200, 201): + error_msg = response.text + raise ProviderException( + f"{self.__class__.__name__} failed to send Gotify message: " + f"HTTP {response.status_code} - {error_msg[:200]}" + ) + + self.logger.info(f"Message sent to Gotify successfully") + return True + + except requests.exceptions.ConnectionError as e: + raise ProviderException( + f"{self.__class__.__name__} failed to connect to Gotify server: {str(e)}" + ) + except requests.exceptions.Timeout: + raise ProviderException( + f"{self.__class__.__name__} connection to Gotify timed out" + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send Gotify message: {str(e)}" + ) + + def test(self): + """Test the Gotify connection by posting a test message.""" + self.logger.debug("Testing Gotify connection") + + if not self.authentication_config: + self.validate_config() + + url = f"{self.authentication_config.server_url}/message?token={self.authentication_config.token}" + + body = { + "message": "KeepHQ test notification - connection successful!", + "title": "KeepHQ Test", + "priority": 0, + } + + try: + response = requests.post( + url, + json=body, + timeout=30, + ) + + if response.status_code in (200, 201): + return { + "ok": True, + "message": "Gotify connection successful - test message sent", + } + elif response.status_code == 401: + return { + "ok": False, + "message": "Gotify connection test failed: Invalid token (401 Unauthorized)", + } + else: + return { + "ok": False, + "message": f"Gotify connection test failed: HTTP {response.status_code} - {response.text[:200]}", + } + + except requests.exceptions.ConnectionError: + return { + "ok": False, + "message": f"Cannot connect to Gotify server {self.authentication_config.server_url}", + } + except Exception as e: + return { + "ok": False, + "message": f"Gotify connection test failed: {str(e)}", + } diff --git a/keep/providers/line_notify_provider/__init__.py b/keep/providers/line_notify_provider/__init__.py new file mode 100644 index 0000000000..662bee3e0b --- /dev/null +++ b/keep/providers/line_notify_provider/__init__.py @@ -0,0 +1 @@ +from .line_notify_provider import LineNotifyProvider diff --git a/keep/providers/line_notify_provider/assets/line.png b/keep/providers/line_notify_provider/assets/line.png new file mode 100644 index 0000000000..e69de29bb2 diff --git a/keep/providers/line_notify_provider/line_notify_provider.py b/keep/providers/line_notify_provider/line_notify_provider.py new file mode 100644 index 0000000000..3a505cc99a --- /dev/null +++ b/keep/providers/line_notify_provider/line_notify_provider.py @@ -0,0 +1,165 @@ +""" +LineNotifyProvider is a class that implements the BaseProvider interface for LINE Notify messages. +Sends alert messages to LINE Notify group or users via the LINE Notify API. + +Note: LINE Notify end of service was March 31, 2025, but API still works. +Migration target: LINE Messaging API (https://developers.line.biz/en/) +""" + +import dataclasses +from typing import Any, Optional + +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig +from keep.validation.fields import HttpsUrl + + +@dataclasses.dataclass +class LineNotifyProviderAuthConfig: + """LINE Notify authentication configuration.""" + + access_token: str = dataclasses.field( + metadata={ + "required": True, + "description": "LINE Notify Access Token", + "sensitive": True, + } + ) + + +class LineNotifyProvider(BaseProvider): + """Send alert message to LINE Notify.""" + + PROVIDER_DISPLAY_NAME = "LINE Notify" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://notify-bot.line.me/" + PROVIDER_DESCRIPTION = "Send notifications to LINE Notify group or users" + IS_TESTABLE = True + + # LINE Notify API endpoint + NOTIFY_URL = "https://notify-api.line.me/api/notify" + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = LineNotifyProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No resources to dispose.""" + pass + + def _notify( + self, + message: str = "", + image_thumbnail: Optional[str] = None, + image_file: Optional[str] = None, + sticker_package_id: Optional[int] = None, + sticker_id: Optional[int] = None, + **kwargs: dict[str, Any], + ): + """ + Send alert message to LINE Notify. + + Args: + message (str): The message to send + image_thumbnail (str): Thumbnail of the image + image_file (str): Path to image file to send + sticker_package_id (int): LINE sticker package ID + sticker_id (int): LINE sticker ID + """ + access_token = self.authentication_config.access_token + + # Build request headers + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/x-www-form-urlencoded", + } + + # Build request payload + payload = {"message": message} + + # Add optional parameters + if image_thumbnail: + payload["imageThumbnail"] = image_thumbnail + if sticker_package_id and sticker_id: + payload["stickerPackageId"] = str(sticker_package_id) + payload["stickerId"] = str(sticker_id) + + # Send request + try: + response = requests.post( + self.NOTIFY_URL, + headers=headers, + data=payload, + timeout=30, + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send LINE notification: {e}" + ) + + # Check response + if not response.ok: + error_detail = response.text + try: + error_detail = response.json().get("message", response.text) + except Exception: + pass + raise ProviderException( + f"{self.__class__.__name__} failed to send LINE notification: {error_detail}" + ) + + result = response.json() + self.logger.debug(f"LINE Notify response: {result}") + return {"status": "ok", "response": result} + + def test(self) -> dict: + """Test the LINE Notify connection.""" + self.validate_config() + message = "🔍 LINE Notify Test - Verbindung getestet!" + try: + result = self._notify(message=message) + return { + "ok": True, + "message": "LINE Notify Verbindung erfolgreich", + "result": result, + } + except ProviderException as e: + return {"ok": False, "message": str(e)} + + +if __name__ == "__main__": + # Output debug messages + import logging + + logging.basicConfig(level=logging.DEBUG, handlers=[logging.StreamHandler()]) + context_manager = ContextManager(tenant_id="singletenant", workflow_id="test") + + import os + + line_access_token = os.environ.get("LINE_NOTIFY_ACCESS_TOKEN") + + if not line_access_token: + print("LINE_NOTIFY_ACCESS_TOKEN environment variable is required") + print("Get one from: https://notify-bot.line.me/my/") + exit(1) + + config = ProviderConfig( + id="line-notify-test", + description="LINE Notify Output Provider", + authentication={"access_token": line_access_token}, + ) + provider = LineNotifyProvider( + context_manager, provider_id="line-notify-test", config=config + ) + provider.notify(message="Hello from Keep + LINE Notify!") + print("✅ Test notification sent to LINE Notify") diff --git a/keep/providers/matrix_provider/__init__.py b/keep/providers/matrix_provider/__init__.py new file mode 100644 index 0000000000..eb888a4f30 --- /dev/null +++ b/keep/providers/matrix_provider/__init__.py @@ -0,0 +1 @@ +from .matrix_provider import MatrixProvider diff --git a/keep/providers/matrix_provider/assets/matrix.svg b/keep/providers/matrix_provider/assets/matrix.svg new file mode 100644 index 0000000000..818e24bfc9 --- /dev/null +++ b/keep/providers/matrix_provider/assets/matrix.svg @@ -0,0 +1,9 @@ + + + + + + + + + diff --git a/keep/providers/matrix_provider/matrix_provider.py b/keep/providers/matrix_provider/matrix_provider.py new file mode 100644 index 0000000000..fcf460f68e --- /dev/null +++ b/keep/providers/matrix_provider/matrix_provider.py @@ -0,0 +1,202 @@ +""" +MatrixProvider is an interface for sending alerts to Matrix rooms. +Matrix is an open decentralized communication protocol. + +API: https://spec.matrix.org/v1.6/client-server-api/ +""" + +import dataclasses +from typing import Optional + +import pydantic +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig + + +@pydantic.dataclasses.dataclass +class MatrixProviderAuthConfig: + """Matrix authentication configuration.""" + + server_url: str = dataclasses.field( + metadata={ + "required": True, + "description": "Matrix server URL (e.g., https://matrix.org)", + } + ) + + access_token: str = dataclasses.field( + metadata={ + "required": True, + "description": "Matrix access token (from login or client)", + "sensitive": True, + } + ) + + room_id: str = dataclasses.field( + metadata={ + "required": True, + "description": "Matrix room ID to send messages to (e.g., !abc123:matrix.org)", + } + ) + + +class MatrixProvider(BaseProvider): + """Send alert messages to Matrix rooms.""" + + PROVIDER_DISPLAY_NAME = "Matrix" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://matrix.org/" + PROVIDER_DESCRIPTION = "Send Keep alerts to Matrix rooms (Element, Synapse, etc.)" + IS_TESTABLE = True + + MATRIX_API = "https://spec.matrix.org/v1.6/client-server-api/" + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = MatrixProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No cleanup needed.""" + pass + + def _get_headers(self): + """Get request headers with auth token.""" + return { + "Authorization": f"Bearer {self.authentication_config.access_token}", + "Content-Type": "application/json", + } + + def _notify( + self, + message: str = "", + msgtype: str = "m.text", + **kwargs: dict, + ): + """ + Send notification message to Matrix room. + + Args: + message (str): Message body to send + msgtype (str): Message type, defaults to "m.text" + """ + self.logger.debug("Sending message to Matrix room") + + if not message: + raise ProviderException( + f"{self.__class__.__name__} failed to send message: message body is required" + ) + + url = f"{self.authentication_config.server_url}/_matrix/client/v3/rooms/{self.authentication_config.room_id}/send/{msgtype}" + + body = { + "body": message, + "msgtype": msgtype, + } + + if msgtype == "m.html" and kwargs.get("formatted_body"): + body["format"] = "org.matrix.html" + body["format_body"] = kwargs.get("formatted_body") + + try: + response = requests.post( + url, + json=body, + headers=self._get_headers(), + timeout=30, + ) + + if response.status_code != 200: + error_msg = response.text + raise ProviderException( + f"{self.__class__.__name__} failed to send Matrix message: " + f"HTTP {response.status_code} - {error_msg[:200]}" + ) + + result = response.json() + self.logger.info( + f"Message sent to Matrix room {self.authentication_config.room_id}, event_id: {result.get('event_id')}" + ) + return True + + except requests.exceptions.ConnectionError as e: + raise ProviderException( + f"{self.__class__.__name__} failed to connect to Matrix server {self.authentication_config.server_url}: {str(e)}" + ) + except requests.exceptions.Timeout: + raise ProviderException( + f"{self.__class__.__name__} connection to Matrix server timed out" + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send Matrix message: {str(e)}" + ) + + def test(self): + """Test the Matrix connection by verifying auth token.""" + self.logger.debug("Testing Matrix connection") + + if not self.authentication_config: + self.validate_config() + + # Try /whoami first, then /sync as fallback + # Matrix servers vary - /whoami works on Synapse, some on /sync + whoami_url = f"{self.authentication_config.server_url}/_matrix/client/r0/whoami" + sync_url = f"{self.authentication_config.server_url}/_matrix/client/r0/sync" + + for test_url, test_name in [(whoami_url, "/whoami"), (sync_url, "/sync")]: + try: + response = requests.get( + test_url, + headers=self._get_headers(), + timeout=30, + ) + + if response.status_code == 200: + if test_name == "/whoami": + user_info = response.json() + user_id = user_info.get('user_id', 'unknown') + return { + "ok": True, + "message": f"Matrix connection successful - user: {user_id}", + } + elif test_name == "/sync": + # /sync 200 means valid auth (even with empty rooms) + return { + "ok": True, + "message": "Matrix connection successful", + } + elif response.status_code == 401: + return { + "ok": False, + "message": "Matrix connection test failed: Invalid access token (401 Unauthorized)", + } + # Try next endpoint + except requests.exceptions.ConnectionError: + if test_name == "/sync": + return { + "ok": False, + "message": f"Cannot connect to Matrix server {self.authentication_config.server_url}", + } + continue + except Exception as e: + if test_name == "/sync": + return { + "ok": False, + "message": f"Matrix connection test failed: {str(e)}", + } + continue + + return { + "ok": False, + "message": "Matrix connection test failed: Server does not respond to auth checks", + } diff --git a/keep/providers/rocketchat_provider/__init__.py b/keep/providers/rocketchat_provider/__init__.py new file mode 100644 index 0000000000..60d8df9a5b --- /dev/null +++ b/keep/providers/rocketchat_provider/__init__.py @@ -0,0 +1,3 @@ +from .rocketchat_provider import RocketChatProvider + +__all__ = ["RocketChatProvider"] diff --git a/keep/providers/rocketchat_provider/rocketchat_provider.py b/keep/providers/rocketchat_provider/rocketchat_provider.py new file mode 100644 index 0000000000..97d83b52ed --- /dev/null +++ b/keep/providers/rocketchat_provider/rocketchat_provider.py @@ -0,0 +1,158 @@ +""" +RocketChatProvider is a notification provider for Rocket.Chat. +API: https://developer.rocket.chat/docs/incoming-webhooks +""" + +import dataclasses +from typing import Optional + +import pydantic +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig + + +@pydantic.dataclasses.dataclass +class RocketChatProviderAuthConfig: + """Rocket.Chat authentication configuration.""" + + webhook_url: str = dataclasses.field( + metadata={ + "required": True, + "description": "Rocket.Chat Incoming Webhook URL (e.g., https://open.rocket.chat/hooks/XXXXX)", + "sensitive": True, + } + ) + channel: str = dataclasses.field( + metadata={ + "required": True, + "description": "Rocket.Chat channel name or ID to send messages to", + } + ) + + +class RocketChatProvider(BaseProvider): + """Send alert messages to Rocket.Chat channels via incoming webhooks.""" + + PROVIDER_DISPLAY_NAME = "Rocket.Chat" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://rocket.chat/" + PROVIDER_DESCRIPTION = "Send Keep alerts to Rocket.Chat channels via incoming webhooks" + IS_TESTABLE = True + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = RocketChatProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No cleanup needed.""" + pass + + def _notify( + self, + message: str = "", + title: str = "", + **kwargs: dict, + ): + """Send notification to Rocket.Chat channel.""" + self.logger.debug("Sending message to Rocket.Chat") + + if not message: + raise ProviderException( + f"{self.__class__.__name__} failed to send message: message body is required" + ) + + display_text = message + if title: + display_text = f"*{title}*\n\n{message}" + + payload = { + "text": display_text, + "channel": self.authentication_config.channel, + } + + headers = {"Content-Type": "application/json"} + + try: + response = requests.post( + self.authentication_config.webhook_url, + json=payload, + headers=headers, + timeout=30, + ) + + if response.status_code not in (200, 201, 204): + error_msg = response.text + raise ProviderException( + f"{self.__class__.__name__} failed to send Rocket.Chat message: " + f"HTTP {response.status_code} - {error_msg[:200]}" + ) + + self.logger.info(f"Message sent to Rocket.Chat channel {self.authentication_config.channel}") + return True + + except requests.exceptions.ConnectionError as e: + raise ProviderException( + f"{self.__class__.__name__} failed to connect to Rocket.Chat: {str(e)}" + ) + except requests.exceptions.Timeout: + raise ProviderException( + f"{self.__class__.__name__} connection to Rocket.Chat timed out" + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send Rocket.Chat message: {str(e)}" + ) + + def test(self): + """Test the Rocket.Chat connection.""" + self.logger.debug("Testing Rocket.Chat connection") + + if not self.authentication_config: + self.validate_config() + + payload = { + "text": "KeepHQ test notification - connection successful!", + "channel": self.authentication_config.channel, + } + + headers = {"Content-Type": "application/json"} + + try: + response = requests.post( + self.authentication_config.webhook_url, + json=payload, + headers=headers, + timeout=30, + ) + + if response.status_code in (200, 201, 204): + return { + "ok": True, + "message": "Rocket.Chat connection successful - test message sent", + } + else: + return { + "ok": False, + "message": f"Rocket.Chat connection test failed: HTTP {response.status_code} - {response.text[:200]}", + } + + except requests.exceptions.ConnectionError: + return { + "ok": False, + "message": "Cannot connect to Rocket.Chat webhook URL", + } + except Exception as e: + return { + "ok": False, + "message": f"Rocket.Chat connection test failed: {str(e)}", + } diff --git a/keep/providers/zulip_provider/__init__.py b/keep/providers/zulip_provider/__init__.py new file mode 100644 index 0000000000..c56bc8b410 --- /dev/null +++ b/keep/providers/zulip_provider/__init__.py @@ -0,0 +1,3 @@ +from .zulip_provider import ZulipProvider + +__all__ = ["ZulipProvider"] diff --git a/keep/providers/zulip_provider/zulip_provider.py b/keep/providers/zulip_provider/zulip_provider.py new file mode 100644 index 0000000000..107952672c --- /dev/null +++ b/keep/providers/zulip_provider/zulip_provider.py @@ -0,0 +1,186 @@ +""" +ZulipProvider is a notification provider for Zulip. +API: https://zulip.com/api/send-message +""" + +import dataclasses +from typing import Optional + +import pydantic +import requests + +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig + + +@pydantic.dataclasses.dataclass +class ZulipProviderAuthConfig: + """Zulip authentication configuration.""" + + url: str = dataclasses.field( + metadata={ + "required": True, + "description": "Zulip server URL (e.g., https://zulip.example.com)", + } + ) + bot_email: str = dataclasses.field( + metadata={ + "required": True, + "description": "Zulip bot email address", + } + ) + api_key: str = dataclasses.field( + metadata={ + "required": True, + "description": "Zulip bot API key", + "sensitive": True, + } + ) + stream: str = dataclasses.field( + metadata={ + "required": True, + "description": "Zulip stream name to send messages to", + } + ) + + +class ZulipProvider(BaseProvider): + """Send alert messages to Zulip streams via bot API.""" + + PROVIDER_DISPLAY_NAME = "Zulip" + PROVIDER_CATEGORY = ["Collaboration"] + PROVIDER_LINK = "https://zulip.com/" + PROVIDER_DESCRIPTION = "Send Keep alerts to Zulip streams via bot API" + IS_TESTABLE = True + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def validate_config(self): + self.authentication_config = ZulipProviderAuthConfig( + **self.config.authentication + ) + + def dispose(self): + """No cleanup needed.""" + pass + + def _notify( + self, + message: str = "", + title: str = "", + **kwargs: dict, + ): + """Send notification to Zulip stream.""" + self.logger.debug("Sending message to Zulip") + + if not message: + raise ProviderException( + f"{self.__class__.__name__} failed to send message: message body is required" + ) + + topic = title if title else "KeepHQ Alert" + display_message = message + if title: + display_message = f"**{title}**\n\n{message}" + + url = f"{self.authentication_config.url}/api/messages" + + payload = { + "type": "stream", + "to": self.authentication_config.stream, + "subject": topic, + "content": display_message, + } + + headers = {"Content-Type": "application/json"} + + try: + response = requests.post( + url, + json=payload, + headers=headers, + auth=(self.authentication_config.bot_email, self.authentication_config.api_key), + timeout=30, + ) + + if response.status_code != 200: + error_msg = response.text + raise ProviderException( + f"{self.__class__.__name__} failed to send Zulip message: " + f"HTTP {response.status_code} - {error_msg[:200]}" + ) + + self.logger.info(f"Message sent to Zulip stream {self.authentication_config.stream}") + return True + + except requests.exceptions.ConnectionError as e: + raise ProviderException( + f"{self.__class__.__name__} failed to connect to Zulip: {str(e)}" + ) + except requests.exceptions.Timeout: + raise ProviderException( + f"{self.__class__.__name__} connection to Zulip timed out" + ) + except requests.exceptions.RequestException as e: + raise ProviderException( + f"{self.__class__.__name__} failed to send Zulip message: {str(e)}" + ) + + def test(self): + """Test the Zulip connection.""" + self.logger.debug("Testing Zulip connection") + + if not self.authentication_config: + self.validate_config() + + url = f"{self.authentication_config.url}/api/messages" + + payload = { + "type": "stream", + "to": self.authentication_config.stream, + "subject": "KeepHQ Test", + "content": "KeepHQ test notification - connection successful!", + } + + headers = {"Content-Type": "application/json"} + + try: + response = requests.post( + url, + json=payload, + headers=headers, + auth=(self.authentication_config.bot_email, self.authentication_config.api_key), + timeout=30, + ) + + if response.status_code == 200: + return { + "ok": True, + "message": "Zulip connection successful - test message sent", + } + elif response.status_code == 401: + return { + "ok": False, + "message": "Zulip connection test failed: Invalid credentials (401 Unauthorized)", + } + else: + return { + "ok": False, + "message": f"Zulip connection test failed: HTTP {response.status_code} - {response.text[:200]}", + } + + except requests.exceptions.ConnectionError: + return { + "ok": False, + "message": f"Cannot connect to Zulip server {self.authentication_config.url}", + } + except Exception as e: + return { + "ok": False, + "message": f"Zulip connection test failed: {str(e)}", + }