# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import re
import requests
from json import loads
from json import dumps
from itertools import chain
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
from ..common import NotifyImageSize
from ..common import NotifyFormat
from ..common import NotifyType
from ..utils import parse_list
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
IS_CHANNEL = re.compile(r'^#(?P<name>[A-Za-z0-9_-]+)$')
IS_USER = re.compile(r'^@(?P<name>[A-Za-z0-9._-]+)$')
IS_ROOM_ID = re.compile(r'^(?P<name>[A-Za-z0-9]+)$')
# Extend HTTP Error Messages
RC_HTTP_ERROR_MAP = {
400: 'Channel/RoomId is wrong format, or missing from server.',
401: 'Authentication tokens provided is invalid or missing.',
}
class RocketChatAuthMode:
"""
The Chat Authentication mode is detected
"""
# providing a webhook
WEBHOOK = "webhook"
# Support token submission
TOKEN = "token"
# Providing a username and password (default)
BASIC = "basic"
# Define our authentication modes
ROCKETCHAT_AUTH_MODES = (
RocketChatAuthMode.WEBHOOK,
RocketChatAuthMode.TOKEN,
RocketChatAuthMode.BASIC,
)
class NotifyRocketChat(NotifyBase):
"""
A wrapper for Notify Rocket.Chat Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Rocket.Chat'
# The services URL
service_url = 'https://rocket.chat/'
# The default protocol
protocol = 'rocket'
# The default secure protocol
secure_protocol = 'rockets'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_rocketchat'
# Allows the user to specify the NotifyImageSize object; this is supported
# through the webhook
image_size = NotifyImageSize.XY_128
# The title is not used
title_maxlen = 0
# The maximum size of the message
body_maxlen = 1000
# Default to markdown
notify_format = NotifyFormat.MARKDOWN
# Define object templates
templates = (
'{schema}://{user}:{password}@{host}:{port}/{targets}',
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{user}:{token}@{host}:{port}/{targets}',
'{schema}://{user}:{token}@{host}/{targets}',
'{schema}://{webhook}@{host}',
'{schema}://{webhook}@{host}:{port}',
'{schema}://{webhook}@{host}/{targets}',
'{schema}://{webhook}@{host}:{port}/{targets}',
)
# Define our template arguments
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
'token': {
'name': _('API Token'),
'map_to': 'password',
'private': True,
},
'webhook': {
'name': _('Webhook'),
'type': 'string',
},
'target_channel': {
'name': _('Target Channel'),
'type': 'string',
'prefix': '#',
'map_to': 'targets',
},
'target_user': {
'name': _('Target User'),
'type': 'string',
'prefix': '@',
'map_to': 'targets',
},
'target_room': {
'name': _('Target Room ID'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'mode': {
'name': _('Webhook Mode'),
'type': 'choice:string',
'values': ROCKETCHAT_AUTH_MODES,
},
'avatar': {
'name': _('Use Avatar'),
'type': 'bool',
'default': False,
},
'webhook': {
'alias_of': 'webhook',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, webhook=None, targets=None, mode=None, avatar=None,
**kwargs):
"""
Initialize Notify Rocket.Chat Object
"""
super().__init__(**kwargs)
# Set our schema
self.schema = 'https' if self.secure else 'http'
# Prepare our URL
self.api_url = '%s://%s' % (self.schema, self.host)
if isinstance(self.port, int):
self.api_url += ':%d' % self.port
# Initialize channels list
self.channels = list()
# Initialize room list
self.rooms = list()
# Initialize user list (webhook only)
self.users = list()
# Assign our webhook (if defined)
self.webhook = webhook
# Used to track token headers upon authentication (if successful)
# This is only used if not on webhook mode
self.headers = {}
# Authentication mode
self.mode = None \
if not isinstance(mode, str) \
else mode.lower()
if self.mode and self.mode not in ROCKETCHAT_AUTH_MODES:
msg = 'The authentication mode specified ({}) is invalid.'.format(
mode)
self.logger.warning(msg)
raise TypeError(msg)
# Detect our mode if it wasn't specified
if not self.mode:
if self.webhook is not None:
# Just a username was specified, we treat this as a webhook
self.mode = RocketChatAuthMode.WEBHOOK
elif self.password and len(self.password) > 32:
self.mode = RocketChatAuthMode.TOKEN
else:
self.mode = RocketChatAuthMode.BASIC
self.logger.debug(
"Auto-Detected Rocketchat Auth Mode: %s", self.mode)
if self.mode in (RocketChatAuthMode.BASIC, RocketChatAuthMode.TOKEN) \
and not (self.user and self.password):
# Username & Password is required for Rocket Chat to work
msg = 'No Rocket.Chat {} was specified.'.format(
'user/pass combo' if self.mode == RocketChatAuthMode.BASIC else
'user/apikey')
self.logger.warning(msg)
raise TypeError(msg)
elif self.mode == RocketChatAuthMode.WEBHOOK and not self.webhook:
msg = 'No Rocket.Chat Incoming Webhook was specified.'
self.logger.warning(msg)
raise TypeError(msg)
if self.mode == RocketChatAuthMode.TOKEN:
# Set our headers for further communication
self.headers.update({
'X-User-Id': self.user,
'X-Auth-Token': self.password,
})
# Validate recipients and drop bad ones:
for recipient in parse_list(targets):
result = IS_CHANNEL.match(recipient)
if result:
# store valid device
self.channels.append(result.group('name'))
continue
result = IS_ROOM_ID.match(recipient)
if result:
# store valid room
self.rooms.append(result.group('name'))
continue
result = IS_USER.match(recipient)
if result:
# store valid room
self.users.append(result.group('name'))
continue
self.logger.warning(
'Dropped invalid channel/room/user '
'({}) specified.'.format(recipient),
)
if self.mode == RocketChatAuthMode.BASIC and \
len(self.rooms) == 0 and len(self.channels) == 0:
msg = 'No Rocket.Chat room and/or channels specified to notify.'
self.logger.warning(msg)
raise TypeError(msg)
# Prepare our avatar setting
# - if specified; that trumps all
# - if not specified and we're dealing with a basic setup, the Avatar
# is disabled by default. This is because if the account doesn't
# have the bot flag set on it it won't work as documented here:
# https://developer.rocket.chat/api/rest-api/endpoints\
# /team-collaboration-endpoints/chat/postmessage
# - Otherwise if we're a webhook, we enable the avatar by default
# (if not otherwise specified) since it will work nicely.
# Place an avatar image to associate with our content
if self.mode == RocketChatAuthMode.BASIC:
self.avatar = False if avatar is None else avatar
else: # self.mode == RocketChatAuthMode.WEBHOOK:
self.avatar = True if avatar is None else avatar
return
def url(self, privacy=False, *args, **kwargs):
"""
Returns the URL built dynamically based on specified arguments.
"""
# Define any URL parameters
params = {
'avatar': 'yes' if self.avatar else 'no',
'mode': self.mode,
}
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Determine Authentication
if self.mode in (RocketChatAuthMode.BASIC, RocketChatAuthMode.TOKEN):
auth = '{user}:{password}@'.format(
user=NotifyRocketChat.quote(self.user, safe=''),
password=self.pprint(
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
)
else:
auth = '{user}{webhook}@'.format(
user='{}:'.format(NotifyRocketChat.quote(self.user, safe=''))
if self.user else '',
webhook=self.pprint(self.webhook, privacy,
mode=PrivacyMode.Secret, safe=''),
)
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}/{targets}/?{params}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
# never encode hostname since we're expecting it to be a valid one
hostname=self.host,
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
targets='/'.join(
[NotifyRocketChat.quote(x, safe='@#') for x in chain(
# Channels are prefixed with a pound/hashtag symbol
['#{}'.format(x) for x in self.channels],
# Rooms are as is
self.rooms,
# Users
['@{}'.format(x) for x in self.users],
)]),
params=NotifyRocketChat.urlencode(params),
)
def __len__(self):
"""
Returns the number of targets associated with this notification
"""
targets = len(self.channels) + len(self.rooms) + len(self.users)
return targets if targets > 0 else 1
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
wrapper to _send since we can alert more then one channel
"""
# Call the _send_ function applicable to whatever mode we're in
# - calls _send_webhook_notification if the mode variable is set
# - calls _send_basic_notification if the mode variable is not set
return getattr(self, '_send_{}_notification'.format(
RocketChatAuthMode.WEBHOOK
if self.mode == RocketChatAuthMode.WEBHOOK
else RocketChatAuthMode.BASIC))(
body=body, title=title, notify_type=notify_type, **kwargs)
def _send_webhook_notification(self, body, title='',
notify_type=NotifyType.INFO, **kwargs):
"""
Sends a webhook notification
"""
# Our payload object
payload = self._payload(body, title, notify_type)
# Assemble our webhook URL
path = 'hooks/{}'.format(self.webhook)
# Build our list of channels/rooms/users (if any identified)
targets = ['@{}'.format(u) for u in self.users]
targets.extend(['#{}'.format(c) for c in self.channels])
targets.extend(['{}'.format(r) for r in self.rooms])
if len(targets) == 0:
# We can take an early exit
return self._send(
payload, notify_type=notify_type, path=path, **kwargs)
# Otherwise we want to iterate over each of the targets
# Initiaize our error tracking
has_error = False
while len(targets):
# Retrieve our target
target = targets.pop(0)
# Assign our channel/room/user
payload['channel'] = target
if not self._send(
payload, notify_type=notify_type, path=path, **kwargs):
# toggle flag
has_error = True
return not has_error
def _send_basic_notification(self, body, title='',
notify_type=NotifyType.INFO, **kwargs):
"""
Authenticates with the server using a user/pass combo for
notifications.
"""
# Track whether we authenticated okay
if self.mode == RocketChatAuthMode.BASIC and not self.login():
return False
# prepare JSON Object
_payload = self._payload(body, title, notify_type)
# Initiaize our error tracking
has_error = False
# Build our list of channels/rooms/users (if any identified)
channels = ['@{}'.format(u) for u in self.users]
channels.extend(['#{}'.format(c) for c in self.channels])
# Create a copy of our channels to notify against
payload = _payload.copy()
while len(channels) > 0:
# Get Channel
channel = channels.pop(0)
payload['channel'] = channel
if not self._send(payload, notify_type=notify_type, **kwargs):
# toggle flag
has_error = True
# Create a copy of our room id's to notify against
rooms = list(self.rooms)
payload = _payload.copy()
while len(rooms):
# Get Room
room = rooms.pop(0)
payload['roomId'] = room
if not self._send(payload, notify_type=notify_type, **kwargs):
# toggle flag
has_error = True
if self.mode == RocketChatAuthMode.BASIC:
# logout
self.logout()
return not has_error
def _payload(self, body, title='', notify_type=NotifyType.INFO):
"""
Prepares a payload object
"""
# prepare JSON Object
payload = {
"text": body,
}
# apply our images if they're set to be displayed
image_url = self.image_url(notify_type)
if self.avatar and image_url:
payload['avatar'] = image_url
return payload
def _send(self, payload, notify_type, path='api/v1/chat.postMessage',
**kwargs):
"""
Perform Notify Rocket.Chat Notification
"""
api_url = '{}/{}'.format(self.api_url, path)
self.logger.debug('Rocket.Chat POST URL: %s (cert_verify=%r)' % (
api_url, self.verify_certificate))
self.logger.debug('Rocket.Chat Payload: %s' % str(payload))
# Copy our existing headers
headers = self.headers.copy()
# Apply minimum headers
headers.update({
'User-Agent': self.app_id,
'Content-Type': 'application/json',
})
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
api_url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyRocketChat.http_response_code_lookup(
r.status_code, RC_HTTP_ERROR_MAP)
self.logger.warning(
'Failed to send Rocket.Chat {}:notification: '
'{}{}error={}.'.format(
self.mode,
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug('Response Details:\r\n{}'.format(r.content))
# Return; we're done
return False
else:
self.logger.info(
'Sent Rocket.Chat {}:notification.'.format(self.mode))
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending Rocket.Chat '
'{}:notification.'.format(self.mode))
self.logger.debug('Socket Exception: %s' % str(e))
# Return; we're done
return False
return True
def login(self):
"""
login to our server
"""
payload = {
'username': self.user,
'password': self.password,
}
api_url = '{}/{}'.format(self.api_url, 'api/v1/login')
try:
r = requests.post(
api_url,
data=payload,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyRocketChat.http_response_code_lookup(
r.status_code, RC_HTTP_ERROR_MAP)
self.logger.warning(
'Failed to authenticate {} with Rocket.Chat: '
'{}{}error={}.'.format(
self.user,
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug('Response Details:\r\n{}'.format(r.content))
# Return; we're done
return False
else:
self.logger.debug('Rocket.Chat authentication successful')
response = loads(r.content)
if response.get('status') != "success":
self.logger.warning(
'Could not authenticate {} with Rocket.Chat.'.format(
self.user))
return False
# Set our headers for further communication
self.headers['X-Auth-Token'] = response.get(
'data', {'authToken': None}).get('authToken')
self.headers['X-User-Id'] = response.get(
'data', {'userId': None}).get('userId')
except (AttributeError, TypeError, ValueError):
# Our response was not the JSON type we had expected it to be
# - ValueError = r.content is Unparsable
# - TypeError = r.content is None
# - AttributeError = r is None
self.logger.warning(
'A commuication error occurred authenticating {} on '
'Rocket.Chat.'.format(self.user))
return False
except requests.RequestException as e:
self.logger.warning(
'A connection error occurred authenticating {} on '
'Rocket.Chat.'.format(self.user))
self.logger.debug('Socket Exception: %s' % str(e))
return False
return True
def logout(self):
"""
logout of our server
"""
api_url = '{}/{}'.format(self.api_url, 'api/v1/logout')
try:
r = requests.post(
api_url,
headers=self.headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyRocketChat.http_response_code_lookup(
r.status_code, RC_HTTP_ERROR_MAP)
self.logger.warning(
'Failed to logoff {} from Rocket.Chat: '
'{}{}error={}.'.format(
self.user,
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug('Response Details:\r\n{}'.format(r.content))
# Return; we're done
return False
else:
self.logger.debug(
'Rocket.Chat log off successful; response %s.' % (
r.content))
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred logging off the '
'Rocket.Chat server')
self.logger.debug('Socket Exception: %s' % str(e))
return False
return True
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to re-instantiate this object.
"""
try:
# Attempt to detect the webhook (if specified in the URL)
# If no webhook is specified, then we just pass along as if nothing
# happened. However if we do find a webhook, we want to rebuild our
# URL without it since it conflicts with standard URLs. Support
# %2F since that is a forward slash escaped
# rocket://webhook@host
# rocket://user:webhook@host
match = re.match(
r'^\s*(?P<schema>[^:]+://)((?P<user>[^:]+):)?'
r'(?P<webhook>[a-z0-9]+(/|%2F)'
r'[a-z0-9]+)\@(?P<url>.+)$', url, re.I)
except TypeError:
# Not a string
return None
if match:
# Re-assemble our URL without the webhook
url = '{schema}{user}{url}'.format(
schema=match.group('schema'),
user='{}@'.format(match.group('user'))
if match.group('user') else '',
url=match.group('url'),
)
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
if match:
# store our webhook
results['webhook'] = \
NotifyRocketChat.unquote(match.group('webhook'))
# Take on the password too in the event we're in basic mode
# We do not unquote() as this is done at a later state
results['password'] = match.group('webhook')
# Apply our targets
results['targets'] = NotifyRocketChat.split_path(results['fullpath'])
# The user may have forced the mode
if 'mode' in results['qsd'] and len(results['qsd']['mode']):
results['mode'] = \
NotifyRocketChat.unquote(results['qsd']['mode'])
# avatar icon
if 'avatar' in results['qsd'] and len(results['qsd']['avatar']):
results['avatar'] = parse_bool(results['qsd'].get('avatar', True))
# The 'to' makes it easier to use yaml configuration
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifyRocketChat.parse_list(results['qsd']['to'])
# The 'webhook' over-ride (if specified)
if 'webhook' in results['qsd'] and len(results['qsd']['webhook']):
results['webhook'] = \
NotifyRocketChat.unquote(results['qsd']['webhook'])
return results