mirror/bazarr
1
0
Fork 0
mirror of https://github.com/morpheus65535/bazarr.git synced 2026-07-31 18:17:54 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
bazarr/libs/apprise/plugins/slack.py
morpheus65535 7ed6608567 Updated multiple vendored modules
2026-06-20 00:29:12 -04:00

1764 lines
63 KiB
Python
Raw Permalink Blame History

# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2026, 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.
# There are 2 ways to use this plugin...
# Method 1: Via Webhook:
# Visit https://my.slack.com/services/new/incoming-webhook/
# to create a new incoming webhook for your account. You'll need to
# follow the wizard to pre-determine the channel(s) you want your
# message to broadcast to, and when you're complete, you will
# recieve a URL that looks something like this:
# https://hooks.slack.com/services/T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7
# ^ ^ ^
# | | |
# These are important <--------------^---------^---------------^
#
# Method 2: Via a Bot:
# 1. visit: https://api.slack.com/apps?new_app=1
# 2. Pick an App Name (such as Apprise) and select your workspace. Then
# press 'Create App'
# 3. You'll be able to click on 'Bots' from here where you can then choose
# to add a 'Bot User'. Give it a name and choose 'Add Bot User'.
# 4. Now you can choose 'Install App' to which you can choose 'Install App
# to Workspace'.
# 5. You will need to authorize the app which you get prompted to do.
# 6. Finally you'll get some important information providing you your
# 'OAuth Access Token' and 'Bot User OAuth Access Token' such as:
# slack://{Oauth Access Token}
#
# ... which might look something like:
# slack://xoxp-1234-1234-1234-4ddbc191d40ee098cbaae6f3523ada2d
# ... or:
# slack://xoxb-1234-1234-4ddbc191d40ee098cbaae6f3523ada2d
#
# You must at least give your bot the following access for it to
# be useful:
# - chat:write - MUST be set otherwise you can not post into
# a channel
# - users:read.email - Required if you want to be able to lookup
# users by their email address.
#
# The easiest way to bring a bot into a channel (so that it can send
# a message to it is to invite it. At this time Apprise does not support
# an auto-join functionality. To do this:
# - In the 'Details' section of your channel
# - Click on the 'More' [...] (elipse icon)
# - Click 'Add apps'
# - You will be able to select the Bot App you previously created
# - Your bot will join your channel.
import contextlib
from json import dumps, loads
from json.decoder import JSONDecodeError
import re
from time import time
import requests
from ..apprise_attachment import AppriseAttachment
from ..common import NotifyFormat, NotifyImageSize, NotifyType
from ..locale import gettext_lazy as _
from ..utils.parse import is_email, parse_bool, parse_list, validate_regex
from ..utils.templates import TemplateType, apply_template
from .base import NotifyBase
# Extend HTTP Error Messages
SLACK_HTTP_ERROR_MAP = {
401: "Unauthorized - Invalid Token.",
}
# Used to break path apart into list of channels
CHANNEL_LIST_DELIM = re.compile(r"[ \t\r\n,#\\/]+")
# Channel Regular Expression Parsing
CHANNEL_RE = re.compile(
r"^(?P<channel>[+#@]?[A-Z0-9_-]{1,32})(:(?P<thread_ts>[0-9.]+))?$", re.I
)
class SlackMode:
"""Tracks the mode of which we're using Slack."""
# We're dealing with a webhook
# Our token looks like: T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7
WEBHOOK = "hook"
# Government Webhook
# Our token still looks like: T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7
# however we have a different URL we post to
WEBHOOK_GOV = "gov-hook"
# We're dealing with a bot (using the OAuth Access Token)
# Our token looks like: xoxp-1234-1234-1234-abc124 or
# Our token looks like: xoxb-1234-1234-abc124 or
BOT = "bot"
# Slack Workflow Builder webhook
# URL: https://hooks.slack.com/workflows/T.../F.../X.../Y...
WORKFLOW = "workflow"
# Slack Workflow trigger webhook (newer format)
# URL: https://hooks.slack.com/triggers/T.../X.../Y...
WORKFLOW_TRIGGER = "trigger"
# Define our Slack Modes
SLACK_MODES = (
SlackMode.WEBHOOK,
SlackMode.WEBHOOK_GOV,
SlackMode.BOT,
SlackMode.WORKFLOW,
SlackMode.WORKFLOW_TRIGGER,
)
class NotifySlack(NotifyBase):
"""A wrapper for Slack Notifications."""
# The default descriptive name associated with the Notification
service_name = "Slack"
# The services URL
service_url = "https://slack.com/"
# The default secure protocol
secure_protocol = "slack"
# Allow 50 requests per minute (Tier 2).
# 60/50 = 0.2
request_rate_per_sec = 1.2
# A URL that takes you to the setup/help of the specific protocol
setup_url = "https://appriseit.com/services/slack/"
# Support attachments
attachment_support = True
# The maximum targets to include when doing batch transfers
# Slack Webhook URLs
webhook_url = "https://hooks.slack.com/services"
webhook_gov_url = "https://hooks.slack-gov.com/services"
# Slack Workflow Builder webhook URL bases
# See: https://slack.com/help/articles/360041352714
workflow_url = "https://hooks.slack.com/workflows"
workflow_trigger_url = "https://hooks.slack.com/triggers"
# Slack API URL (used with Bots)
api_url = "https://slack.com/api/{}"
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_72
# The maximum allowable characters allowed in the body per message
body_maxlen = 35000
# There is no reason we should exceed 35KB when reading in a JSON file.
# If it is more than this, then it is not accepted
max_slack_template_size = 35000
# Default Notification Format
notify_format = NotifyFormat.MARKDOWN
# Bot's do not have default channels to notify; so #general
# becomes the default channel in BOT mode
default_notification_channel = "#general"
# Define object templates
templates = (
# Webhook
"{schema}://{token_a}/{token_b}/{token_c}",
"{schema}://{botname}@{token_a}/{token_b}/{token_c}",
"{schema}://{token_a}/{token_b}/{token_c}/{targets}",
"{schema}://{botname}@{token_a}/{token_b}/{token_c}/{targets}",
# Bot
"{schema}://{access_token}/",
"{schema}://{access_token}/{targets}",
)
# Define our template tokens
template_tokens = dict(
NotifyBase.template_tokens,
**{
"botname": {
"name": _("Bot Name"),
"type": "string",
"map_to": "user",
},
# Bot User OAuth Access Token
# which always starts with xoxp- e.g.:
# xoxb-1234-1234-4ddbc191d40ee098cbaae6f3523ada2d
"access_token": {
"name": _("OAuth Access Token"),
"type": "string",
"private": True,
"required": True,
"regex": (r"^(?:xoxe\.)?xox[abp]-[A-Z0-9-]+$", "i"),
},
# Token required as part of the Webhook request
# /AAAAAAAAA/........./........................
"token_a": {
"name": _("Token A"),
"type": "string",
"private": True,
"required": True,
"regex": (r"^[A-Z0-9]+$", "i"),
},
# Token required as part of the Webhook request
# /........./BBBBBBBBB/........................
"token_b": {
"name": _("Token B"),
"type": "string",
"private": True,
"required": True,
"regex": (r"^[A-Z0-9]+$", "i"),
},
# Token required as part of the Webhook request
# /........./........./CCCCCCCCCCCCCCCCCCCCCCCC
"token_c": {
"name": _("Token C"),
"type": "string",
"private": True,
"required": True,
"regex": (r"^[A-Za-z0-9]+$", "i"),
},
"target_encoded_id": {
"name": _("Target Encoded ID"),
"type": "string",
"prefix": "+",
"map_to": "targets",
},
"target_email": {
"name": _("Target Email"),
"type": "string",
"map_to": "targets",
},
"target_user": {
"name": _("Target User"),
"type": "string",
"prefix": "@",
"map_to": "targets",
},
"target_channels": {
"name": _("Target Channel"),
"type": "string",
"prefix": "#",
"map_to": "targets",
},
"targets": {
"name": _("Targets"),
"type": "list:string",
},
# Workflow Builder path: all URL segments joined with slashes
# e.g. T05XXXX/Ft07XXXX/XXXXXXXX/YYYYYYYY
"workflow_path": {
"name": _("Workflow Path"),
"type": "string",
"private": True,
},
},
)
# Define our template arguments
template_args = dict(
NotifyBase.template_args,
**{
"image": {
"name": _("Include Image"),
"type": "bool",
"default": True,
"map_to": "include_image",
},
"footer": {
"name": _("Include Footer"),
"type": "bool",
"default": True,
"map_to": "include_footer",
},
# Use Payload in Blocks (vs legacy way):
# See: https://api.slack.com/reference/messaging/payload
"blocks": {
"name": _("Use Blocks"),
"type": "bool",
"default": False,
"map_to": "use_blocks",
},
"timestamp": {
"name": _("Include Timestamp"),
"type": "bool",
"default": True,
"map_to": "include_timestamp",
},
"mode": {
"name": _("Message Mode"),
"type": "choice:string",
"values": SLACK_MODES,
# mode is detected if not specified
},
"token": {
"name": _("Token"),
"alias_of": ("access_token", "token_a", "token_b", "token_c"),
},
"to": {
"alias_of": "targets",
},
"template": {
"name": _("Template Path"),
"type": "string",
"private": True,
},
},
)
# Define our token control
template_kwargs = {
"tokens": {
"name": _("Template Tokens"),
"prefix": ":",
},
}
# Formatting requirements are defined here:
# https://api.slack.com/docs/message-formatting
_re_formatting_map = {
# New lines must become the string version
r"\r\*\n": "\\n",
# Escape other special characters
r"&": "&amp;",
r"<": "&lt;",
r">": "&gt;",
}
# To notify a channel, one uses <!channel|channel>
_re_channel_support = re.compile(
r"(?P<match>(?:<|\&lt;)?[ \t]*"
r"!(?P<channel>[^| \n]+)"
r"(?:[ \t]*\|[ \t]*(?:(?P<val>[^\n]+?)[ \t]*)?(?:>|\&gt;)"
r"|(?:>|\&gt;)))",
re.IGNORECASE,
)
# To notify a user by their ID, one uses <@U6TTX1F9R>
_re_user_id_support = re.compile(
r"(?P<match>(?:<|\&lt;)?[ \t]*"
r"@(?P<userid>[^| \n]+)"
r"(?:[ \t]*\|[ \t]*(?:(?P<val>[^\n]+?)[ \t]*)?(?:>|\&gt;)"
r"|(?:>|\&gt;)))",
re.IGNORECASE,
)
# The markdown in slack isn't [desc](url), it's <url|desc>
#
# To accommodate this, we need to ensure we don't escape URLs that match
_re_url_support = re.compile(
r"(?P<match>(?:<|\&lt;)?[ \t]*"
r"(?P<url>(?:https?|mailto)://[^| \n]+)"
r"(?:[ \t]*\|[ \t]*(?:(?P<val>[^\n]+?)[ \t]*)?(?:>|\&gt;)"
r"|(?:>|\&gt;)))",
re.IGNORECASE,
)
def __init__(
self,
access_token=None,
token_a=None,
token_b=None,
token_c=None,
targets=None,
include_image=None,
include_footer=None,
include_timestamp=None,
use_blocks=None,
mode=None,
template=None,
tokens=None,
workflow_path=None,
**kwargs,
):
"""Initialize Slack Object."""
super().__init__(**kwargs)
# Store our webhook mode
# Track whether the caller supplied an explicit mode so we can
# decide later whether to validate exact segment count or
# auto-detect the workflow sub-mode from segment count.
_mode_explicit = bool(mode and isinstance(mode, str))
if _mode_explicit:
self.mode = next(
(a for a in SLACK_MODES if a.startswith(mode)), None
)
if self.mode not in SLACK_MODES:
msg = f"The Slack mode specified ({mode}) is invalid."
self.logger.warning(msg)
raise TypeError(msg)
elif workflow_path:
# Mode will be determined by segment count below;
# use WORKFLOW as a sentinel so we enter the right branch.
self.mode = SlackMode.WORKFLOW
else: # Detect
self.mode = SlackMode.BOT if access_token else SlackMode.WEBHOOK
if self.mode in (
SlackMode.WORKFLOW,
SlackMode.WORKFLOW_TRIGGER,
):
# Workflow Builder mode: no OAuth token, no webhook tokens
self.access_token = None
self.token_a = None
self.token_b = None
self.token_c = None
# Parse workflow path segments from string or list
if isinstance(workflow_path, (list, tuple)):
self.workflow_path = [str(s) for s in workflow_path if s]
elif isinstance(workflow_path, str):
self.workflow_path = [s for s in workflow_path.split("/") if s]
else:
self.workflow_path = []
# Segment counts are fixed by Slack:
# /workflows/ URLs have exactly 4 segments
# /triggers/ URLs have exactly 3 segments
seg_count = len(self.workflow_path)
if _mode_explicit:
# Validate exact count for the declared mode
expected = 4 if self.mode == SlackMode.WORKFLOW else 3
if seg_count != expected:
msg = (
f"A Slack {self.mode!r} URL requires exactly"
f" {expected} path segments"
f" ({workflow_path!r} has {seg_count})."
)
self.logger.warning(msg)
raise TypeError(msg)
else:
# Auto-detect sub-mode from segment count
if seg_count == 4:
self.mode = SlackMode.WORKFLOW
elif seg_count == 3:
self.mode = SlackMode.WORKFLOW_TRIGGER
else:
msg = (
"A Slack Workflow URL requires exactly 3 or 4"
f" path segments ({workflow_path!r} has"
f" {seg_count})."
)
self.logger.warning(msg)
raise TypeError(msg)
elif self.mode in (SlackMode.WEBHOOK, SlackMode.WEBHOOK_GOV):
self.workflow_path = []
self.access_token = None
self.token_a = validate_regex(
token_a, *self.template_tokens["token_a"]["regex"]
)
if not self.token_a:
msg = (
"An invalid Slack (first) Token "
f"({token_a}) was specified."
)
self.logger.warning(msg)
raise TypeError(msg)
self.token_b = validate_regex(
token_b, *self.template_tokens["token_b"]["regex"]
)
if not self.token_b:
msg = (
"An invalid Slack (second) Token "
f"({token_b}) was specified."
)
self.logger.warning(msg)
raise TypeError(msg)
self.token_c = validate_regex(
token_c, *self.template_tokens["token_c"]["regex"]
)
if not self.token_c:
msg = (
"An invalid Slack (third) Token "
f"({token_c}) was specified."
)
self.logger.warning(msg)
raise TypeError(msg)
else:
self.workflow_path = []
self.token_a = None
self.token_b = None
self.token_c = None
self.access_token = validate_regex(
access_token, *self.template_tokens["access_token"]["regex"]
)
if not self.access_token:
msg = (
"An invalid Slack OAuth Access Token "
f"({access_token}) was specified."
)
self.logger.warning(msg)
raise TypeError(msg)
# Look the users up by their email address and map them back to their
# id here for future queries (if needed). This allows people to
# specify a full email as a recipient via slack
self._lookup_users = {}
self.use_blocks = (
parse_bool(use_blocks, self.template_args["blocks"]["default"])
if use_blocks is not None
else self.template_args["blocks"]["default"]
)
# Build list of channels
self.channels = parse_list(targets)
if len(self.channels) == 0:
# No problem; the webhook is smart enough to just notify the
# channel it was created for; adding 'None' is just used as
# a flag lower to not set the channels
self.channels.append(
None
if self.mode
in (
SlackMode.WEBHOOK,
SlackMode.WEBHOOK_GOV,
# Workflow mode has no channel concept
SlackMode.WORKFLOW,
SlackMode.WORKFLOW_TRIGGER,
)
else self.default_notification_channel
)
# Iterate over above list and store content accordingly
self._re_formatting_rules = re.compile(
r"(" + "|".join(self._re_formatting_map.keys()) + r")",
re.IGNORECASE,
)
# Place a thumbnail image inline with the message body
self.include_image = (
self.template_args["image"]["default"]
if include_image is None
else include_image
)
# Place a footer with each post
self.include_footer = (
self.template_args["footer"]["default"]
if include_footer is None
else include_footer
)
# timestamp inclusion (only applicable if footer also defined
self.include_timestamp = (
self.template_args["timestamp"]["default"]
if include_timestamp is None
else include_timestamp
)
# Our template object is just an AppriseAttachment object
self.template = AppriseAttachment(asset=self.asset)
if template:
# Add our definition to our template
self.template.add(template)
if not len(self.template):
# add() failed (unsupported schema, unparseable URL, etc.)
msg = f"The Slack template ({template!r}) could not be loaded."
self.logger.warning(msg)
raise TypeError(msg)
# Enforce maximum file size
self.template[0].max_file_size = self.max_slack_template_size
# Template functionality
self.tokens = {}
if isinstance(tokens, dict):
self.tokens.update(tokens)
elif tokens:
msg = (
"The specified Slack Template Tokens "
f"({tokens}) are not identified as a dictionary."
)
self.logger.warning(msg)
raise TypeError(msg)
# A template always implies Block Kit mode; there is no meaningful
# use for templates outside of blocks mode.
if self.template:
self.use_blocks = True
return
def gen_payload(
self, body, title="", notify_type=NotifyType.INFO, **kwargs
):
"""Generate our payload from an external Block Kit JSON template.
Returns the validated attachment content dict (which must contain
a non-empty 'blocks' list with typed entries) to be placed into
the Slack attachments array. Returns False on any failure.
"""
# Acquire the first template attachment
template = self.template[0]
if not template:
# We could not access the attachment
self.logger.error(
"Could not access Slack template"
f" {template.url(privacy=True)}."
)
return False
# Acquire our to-be footer icon if configured to do so
image_url = (
None if not self.include_image else self.image_url(notify_type)
)
# Take a copy of our token dictionary
tokens = self.tokens.copy()
# Apply some defaults template values
tokens["app_body"] = body
tokens["app_title"] = title
tokens["app_type"] = notify_type.value
tokens["app_id"] = self.app_id
tokens["app_desc"] = self.app_desc
tokens["app_color"] = self.color(notify_type)
tokens["app_image_url"] = image_url
tokens["app_url"] = self.app_url
# Templates are always Block Kit JSON; enforce JSON escaping
tokens["app_mode"] = TemplateType.JSON
# Coerce all substitution values to str before JSON-escaping.
# apply_template's _escape_json() calls json.dumps(v)[1:-1] which
# is only correct for strings; None produces "ul" and other non-
# string types produce similarly corrupted output. app_mode is a
# TemplateType sentinel passed as a named parameter, not a
# substitution value, so it is left as-is.
safe_tokens = {
k: (
v
if k == "app_mode" or isinstance(v, str)
else ("" if v is None else str(v))
)
for k, v in tokens.items()
}
try:
with open(template.path) as fp:
content = apply_template(fp.read(), **safe_tokens)
except OSError:
self.logger.error(
"Slack template"
f" {template.url(privacy=True)} could not be read."
)
return False
# Parse and validate as JSON
try:
content = loads(content)
except JSONDecodeError as e:
self.logger.error(
"Slack template"
f" {template.url(privacy=True)} contains invalid JSON."
)
self.logger.debug(f"JSONDecodeError: {e}")
return False
# Template must parse to a JSON object, not an array or scalar
if not isinstance(content, dict):
self.logger.error(
"Slack template"
f" {template.url(privacy=True)} must be a JSON object"
" (got {}).".format(type(content).__name__)
)
return False
# 'blocks' must be a non-empty list (Block Kit requirement)
if (
not isinstance(content.get("blocks"), list)
or not content["blocks"]
):
self.logger.error(
"Slack template"
f" {template.url(privacy=True)} must contain"
" a non-empty 'blocks' list."
)
return False
# Every block must be a dict with a 'type' string (Block Kit)
if not all(
isinstance(b, dict) and isinstance(b.get("type"), str)
for b in content["blocks"]
):
self.logger.error(
"Slack template"
f" {template.url(privacy=True)} contains"
" a block missing a 'type' string."
)
return False
# Return the validated attachment content dict
return content
def send(
self,
body,
title="",
notify_type=NotifyType.INFO,
attach=None,
**kwargs,
):
"""Perform Slack Notification."""
# error tracking (used for function return)
has_error = False
#
# Workflow Builder mode: fixed endpoint, no channel iteration
#
if self.mode in (SlackMode.WORKFLOW, SlackMode.WORKFLOW_TRIGGER):
if self.template:
# Use external Block Kit JSON template
payload = self.gen_payload(
body, title, notify_type=notify_type, **kwargs
)
if payload is False:
return False
else:
# Default payload: combine title and body into a single
# 'text' field -- the workflow must accept this variable.
payload = {
"text": f"{title}: {body}" if title else body,
}
# Construct the workflow POST URL from stored path segments
wf_base = (
self.workflow_trigger_url
if self.mode is SlackMode.WORKFLOW_TRIGGER
else self.workflow_url
)
url = "{}/{}".format(wf_base, "/".join(self.workflow_path))
# _send() returns False on error, non-False on success
return self._send(url, payload) is not False
#
# Prepare JSON Object (applicable to both WEBHOOK and BOT mode)
#
if self.use_blocks:
if self.template:
# Template mode: generate blocks attachment from template
template_content = self.gen_payload(
body, title, notify_type=notify_type, **kwargs
)
if template_content is False:
# Template loading or parsing failed; abort early
return False
# Wrap the template attachment content in a full payload
payload = {
"username": self.user if self.user else self.app_id,
"attachments": [template_content],
}
else:
# Our slack format
slack_format = (
"mrkdwn"
if self.notify_format == NotifyFormat.MARKDOWN
else "plain_text"
)
payload = {
"username": self.user if self.user else self.app_id,
"attachments": [
{
"blocks": [
{
"type": "section",
"text": {
"type": slack_format,
"text": body,
},
}
],
"color": self.color(notify_type),
}
],
}
# Slack only accepts non-empty header sections
if title:
payload["attachments"][0]["blocks"].insert(
0,
{
"type": "header",
"text": {
"type": "plain_text",
"text": title,
"emoji": True,
},
},
)
# Include the footer only if specified to do so
if self.include_footer:
# Acquire our to-be footer icon if configured to do so
image_url = (
None
if not self.include_image
else self.image_url(notify_type)
)
# Prepare our footer based on the block structure
footer = {
"type": "context",
"elements": [
{"type": slack_format, "text": self.app_id}
],
}
if image_url:
payload["icon_url"] = image_url
footer["elements"].insert(
0,
{
"type": "image",
"image_url": image_url,
"alt_text": notify_type,
},
)
payload["attachments"][0]["blocks"].append(footer)
else:
#
# Legacy API Formatting
#
if self.notify_format == NotifyFormat.MARKDOWN:
body = self._re_formatting_rules.sub( # pragma: no branch
lambda x: self._re_formatting_map[x.group()],
body,
)
# Support <!channel|desc>, <!channel> entries
for match in self._re_channel_support.findall(body):
# Swap back any ampersands previously updaated
channel = match[1].strip()
desc = match[2].strip()
# Update our string
body = re.sub(
re.escape(match[0]),
f"<!{channel}|{desc}>" if desc else f"<!{channel}>",
body,
flags=re.IGNORECASE,
)
# Support <@userid|desc>, <@channel> entries
for match in self._re_user_id_support.findall(body):
# Swap back any ampersands previously updaated
user = match[1].strip()
desc = match[2].strip()
# Update our string
body = re.sub(
re.escape(match[0]),
f"<@{user}|{desc}>" if desc else f"<@{user}>",
body,
flags=re.IGNORECASE,
)
# Support <url|desc>, <url> entries
for match in self._re_url_support.findall(body):
# Swap back any ampersands previously updaated
url = match[1].replace("&amp;", "&")
desc = match[2].strip()
# Update our string
body = re.sub(
re.escape(match[0]),
f"<{url}|{desc}>" if desc else f"<{url}>",
body,
flags=re.IGNORECASE,
)
# Perform Formatting on title here; this is not needed for block
# mode above
title = self._re_formatting_rules.sub( # pragma: no branch
lambda x: self._re_formatting_map[x.group()],
title,
)
# Prepare JSON Object (applicable to both WEBHOOK and BOT mode)
payload = {
"username": self.user if self.user else self.app_id,
# Use Markdown language
"mrkdwn": self.notify_format == NotifyFormat.MARKDOWN,
"attachments": [
{
"title": title,
"text": body,
"color": self.color(notify_type),
}
],
}
# Acquire our to-be footer icon if configured to do so
image_url = (
None if not self.include_image else self.image_url(notify_type)
)
if image_url:
payload["icon_url"] = image_url
# Include the footer only if specified to do so
if self.include_footer:
if image_url:
payload["attachments"][0]["footer_icon"] = image_url
# Include the footer only if specified to do so
payload["attachments"][0]["footer"] = self.app_id
if self.include_timestamp:
# Timestamp
payload["attachments"][0]["ts"] = time()
if (
attach
and self.attachment_support
and self.mode in (SlackMode.WEBHOOK, SlackMode.WEBHOOK_GOV)
):
# Be friendly; let the user know why they can't send their
# attachments if using the Webhook mode
self.logger.warning("Slack Webhooks do not support attachments.")
# Prepare our Slack URL (depends on mode)
if self.mode is SlackMode.WEBHOOK:
url = (
f"{self.webhook_url}/{self.token_a}"
f"/{self.token_b}/{self.token_c}"
)
elif self.mode is SlackMode.WEBHOOK_GOV:
url = (
f"{self.webhook_gov_url}/{self.token_a}"
f"/{self.token_b}/{self.token_c}"
)
else: # SlackMode.BOT
url = self.api_url.format("chat.postMessage")
# Create a copy of the channel list
channels = list(self.channels)
attach_channel_list = []
while len(channels):
channel = channels.pop(0)
if channel is not None:
# We'll perform a user lookup if we detect an email
email = is_email(channel)
if email:
payload["channel"] = self.lookup_userid(
email["full_email"]
)
if not payload["channel"]:
# Move along; any notifications/logging would have
# come from lookup_userid()
has_error = True
continue
else: # Channel
result = CHANNEL_RE.match(channel)
if not result:
# Channel over-ride was specified
self.logger.warning(
f"The specified Slack target {channel} is invalid;"
"skipping."
)
# Mark our failure
has_error = True
continue
# Store oure content
channel, thread_ts = (
result.group("channel"),
result.group("thread_ts"),
)
if thread_ts:
payload["thread_ts"] = thread_ts
elif "thread_ts" in payload:
# Handle situations where one channel has a thread_id
# specified, and the next does not. We do not want to
# cary forward the last value specified
del payload["thread_ts"]
if channel[0] == "+":
# Treat as encoded id if prefixed with a +
payload["channel"] = channel[1:]
elif channel[0] == "@":
# Treat @ value 'as is'
payload["channel"] = channel
else:
# Prefix with channel hash tag (if not already)
payload["channel"] = (
channel if channel[0] == "#" else f"#{channel}"
)
response = self._send(url, payload)
if not response:
# Handle any error
has_error = True
continue
# Store the valid channel or chat ID (for DMs) that will
# be accepted by Slack's attachment method later.
if response.get("channel"):
attach_channel_list.append(response.get("channel"))
self.logger.info(
"Sent Slack notification{}.".format(
f" to {channel}" if channel is not None else ""
)
)
if (
attach
and self.attachment_support
and self.mode is SlackMode.BOT
and attach_channel_list
):
# Send our attachments (can only be done in bot mode)
for no, attachment in enumerate(attach, start=1):
# Perform some simple error checking
if not attachment:
# We could not access the attachment
self.logger.error(
"Could not access attachment"
f" {attachment.url(privacy=True)}."
)
return False
self.logger.debug(
f"Posting Slack attachment {attachment.url(privacy=True)}"
)
# Get the URL to which to upload the file.
# https://api.slack.com/methods/files.getUploadURLExternal
params = {
"filename": (
attachment.name
if attachment.name
else f"file{no:03}.dat"
),
"length": len(attachment),
}
url_ = self.api_url.format("files.getUploadURLExternal")
response = self._send(
url_, {}, http_method="get", params=params
)
if not (
response
and response.get("file_id")
and response.get("upload_url")
):
self.logger.error("Could retrieve file upload URL.")
# We failed to get an upload URL, take an early exit
return False
file_id = response.get("file_id")
upload_url = response.get("upload_url")
# Upload file
response = self._send(upload_url, {}, attach=attachment)
# Send file to channels
# https://api.slack.com/methods/files.completeUploadExternal
for channel_id in attach_channel_list:
payload_ = {
"files": [
{
"id": file_id,
"title": attachment.name,
}
],
"channel_id": channel_id,
}
url_ = self.api_url.format("files.completeUploadExternal")
response = self._send(url_, payload_)
# Expected response
# {
# "ok": true,
# "files": [
# {
# "id": "F123ABC456",
# "title": "slack-test"
# }
# ]
# }
if not (response and response.get("files")):
self.logger.error("Failed to send file to channel.")
# We failed to send the file to the channel,
# take an early exit
return False
return not has_error
def lookup_userid(self, email):
"""Takes an email address and attempts to resolve/acquire it's user id
for notification purposes."""
if email in self._lookup_users:
# We're done as entry has already been retrieved
return self._lookup_users[email]
if self.mode is not SlackMode.BOT:
# You can not look up
self.logger.warning(
"Emails can not be resolved to Slack User IDs unless you "
"have a bot configured."
)
return None
lookup_url = self.api_url.format("users.lookupByEmail")
headers = {
"User-Agent": self.app_id,
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": f"Bearer {self.access_token}",
}
# we pass in our email address as the argument
params = {
"email": email,
}
self.logger.debug(
"Slack User Lookup POST URL:"
f" {lookup_url} (cert_verify={self.verify_certificate!r})"
)
self.logger.debug(f"Slack User Lookup Parameters: {params!s}")
# Initialize our HTTP JSON response
response = {"ok": False}
# Initialize our detected user id (also the response to this function)
user_id = None
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.get(
lookup_url,
headers=headers,
params=params,
verify=self.verify_certificate,
timeout=self.request_timeout,
allow_redirects=self.redirects,
)
# Attachment posts return a JSON string
with contextlib.suppress(AttributeError, TypeError, ValueError):
# Load our JSON object if we can
# ValueError = r.content is Unparsable
# TypeError = r.content is None
# AttributeError = r is None
response = loads(r.content)
# We can get a 200 response, but still fail. A failure message
# might look like this (missing bot permissions):
# {
# 'ok': False,
# 'error': 'missing_scope',
# 'needed': 'users:read.email',
# 'provided': 'calls:write,chat:write'
# }
if r.status_code != requests.codes.ok or not (
response and response.get("ok", False)
):
# We had a problem
status_str = NotifySlack.http_response_code_lookup(
r.status_code, SLACK_HTTP_ERROR_MAP
)
self.logger.warning(
"Failed to send Slack User Lookup:{}{}error={}.".format(
status_str, ", " if status_str else "", r.status_code
)
)
self.logger.debug(
"Response Details:\r\n%r", (r.content or b"")[:2000]
)
# Return; we're done
return False
# If we reach here, then we were successful in looking up
# the user. A response generally looks like this:
# {
# 'ok': True,
# 'user': {
# 'id': 'J1ZQB9T9Y',
# 'team_id': 'K1WR6TML2',
# 'name': 'l2g',
# 'deleted': False,
# 'color': '9f69e7',
# 'real_name': 'Chris C',
# 'tz': 'America/New_York',
# 'tz_label': 'Eastern Standard Time',
# 'tz_offset': -18000,
# 'profile': {
# 'title': '',
# 'phone': '',
# 'skype': '',
# 'real_name': 'Chris C',
# 'real_name_normalized':
# 'Chris C',
# 'display_name': 'l2g',
# 'display_name_normalized': 'l2g',
# 'fields': None,
# 'status_text': '',
# 'status_emoji': '',
# 'status_expiration': 0,
# 'avatar_hash': 'g785e9c0ddf6',
# 'email': 'lead2gold@gmail.com',
# 'first_name': 'Chris',
# 'last_name': 'C',
# 'image_24': 'https://secure.gravatar.com/...',
# 'image_32': 'https://secure.gravatar.com/...',
# 'image_48': 'https://secure.gravatar.com/...',
# 'image_72': 'https://secure.gravatar.com/...',
# 'image_192': 'https://secure.gravatar.com/...',
# 'image_512': 'https://secure.gravatar.com/...',
# 'status_text_canonical': '',
# 'team': 'K1WR6TML2'
# },
# 'is_admin': True,
# 'is_owner': True,
# 'is_primary_owner': True,
# 'is_restricted': False,
# 'is_ultra_restricted': False,
# 'is_bot': False,
# 'is_app_user': False,
# 'updated': 1603904274
# }
# }
# We're only interested in the id
user_id = response["user"]["id"]
# Cache it for future
self._lookup_users[email] = user_id
self.logger.info(
"Email %s resolves to the Slack User ID: %s.", email, user_id
)
except requests.RequestException as e:
self.logger.warning(
"A Connection error occurred looking up Slack User.",
)
self.logger.debug(f"Socket Exception: {e!s}")
# Return; we're done
return None
return user_id
def _send(
self,
url,
payload,
attach=None,
http_method="post",
params=None,
**kwargs,
):
"""Wrapper to the requests (post) object."""
self.logger.debug(
f"Slack POST URL: {url} (cert_verify={self.verify_certificate!r})"
)
self.logger.debug(f"Slack Payload: {payload!s}")
headers = {
"User-Agent": self.app_id,
"Accept": "application/json",
}
if not attach:
headers["Content-Type"] = "application/json; charset=utf-8"
if self.mode is SlackMode.BOT:
headers["Authorization"] = f"Bearer {self.access_token}"
# Our response object
response = {"ok": False}
# Always call throttle before any remote server i/o is made
self.throttle()
# Our attachment path (if specified)
files = None
try:
# Open our attachment path if required:
if attach:
files = {
"file": (
attach.name,
# file handle is safely closed in `finally`; inline
# open is intentional; attach.open() dispatches to
# BytesIO for memory attachments
attach.open(),
),
}
r = requests.request(
http_method,
url,
data=payload if attach else dumps(payload),
headers=headers,
files=files,
verify=self.verify_certificate,
timeout=self.request_timeout,
allow_redirects=self.redirects,
params=params if params else None,
)
# Posts return a JSON string
with contextlib.suppress(AttributeError, TypeError, ValueError):
# Load our JSON object if we can
# ValueError = r.content is Unparsable
# TypeError = r.content is None
# AttributeError = r is None
response = loads(r.content)
# Another response type is:
# {
# 'ok': False,
# 'error': 'not_in_channel',
# }
status_okay = False
if self.mode is SlackMode.BOT:
status_okay = (
(response and response.get("ok", False))
or
# Responses for file uploads look like this
# 'OK - <file length>'
(
r.content
and isinstance(r.content, bytes)
and b"OK" in r.content
)
)
elif self.mode in (
SlackMode.WORKFLOW,
SlackMode.WORKFLOW_TRIGGER,
):
# Workflow trigger webhooks confirm success via HTTP 200;
# the response body may be empty or non-standard JSON.
status_okay = r.status_code == requests.codes.ok
elif r.content == b"ok":
# The text 'ok' is returned if this is a Webhook request
# So the below captures that as well.
status_okay = True
if r.status_code != requests.codes.ok or not status_okay:
# We had a problem
status_str = NotifySlack.http_response_code_lookup(
r.status_code, SLACK_HTTP_ERROR_MAP
)
self.logger.warning(
"Failed to send{} to Slack: {}{}error={}.".format(
(" " + attach.name) if attach else "",
status_str,
", " if status_str else "",
r.status_code,
)
)
self.logger.debug(f"Response Details:\r\n{r.content}")
return False
# Message Post Response looks like this:
# {
# "attachments": [
# {
# "color": "3AA3E3",
# "fallback": "test",
# "id": 1,
# "text": "my body",
# "title": "my title",
# "ts": 1573694687
# }
# ],
# "bot_id": "BAK4K23G5",
# "icons": {
# "image_48": "https://s3-us-west-2.amazonaws.com/...
# },
# "subtype": "bot_message",
# "text": "",
# "ts": "1573694689.003700",
# "type": "message",
# "username": "Apprise"
# }
# files.completeUploadExternal responses look like this:
# {
# "ok": true,
# "files": [
# {
# "id": "F123ABC456",
# "title": "slack-test"
# }
# ]
# }
except requests.RequestException as e:
self.logger.warning(
"A Connection error occurred posting {}to Slack.".format(
attach.name if attach else ""
)
)
self.logger.debug(f"Socket Exception: {e!s}")
return False
except OSError as e:
self.logger.warning(
"An I/O error occurred while reading {}.".format(
attach.name if attach else "attachment"
)
)
self.logger.debug(f"I/O Exception: {e!s}")
return False
finally:
# Close our file (if it's open) stored in the second element
# of our files tuple (index 1)
if files:
files["file"][1].close()
# Return the response for processing
return response
@property
def url_identifier(self):
"""Returns all of the identifiers that make this URL unique from
another simliar one.
Targets or end points should never be identified here.
"""
if self.mode in (SlackMode.WORKFLOW, SlackMode.WORKFLOW_TRIGGER):
# Workflow mode is identified by its path segments
return (self.secure_protocol, self.mode, *self.workflow_path)
return (
self.secure_protocol,
self.token_a,
self.token_b,
self.token_c,
self.access_token,
)
def url(self, privacy=False, *args, **kwargs):
"""Returns the URL built dynamically based on specified arguments."""
# Define any URL parameters
params = {
"image": "yes" if self.include_image else "no",
"footer": "yes" if self.include_footer else "no",
"timestamp": "yes" if self.include_timestamp else "no",
"blocks": "yes" if self.use_blocks else "no",
"mode": self.mode,
}
if self.template:
params["template"] = NotifySlack.quote(
self.template[0].url(), safe=""
)
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
# Store any template token entries if specified
params.update({f":{k}": v for k, v in self.tokens.items()})
# Determine if there is a botname present
botname = ""
if self.user:
botname = "{botname}@".format(
botname=NotifySlack.quote(self.user, safe=""),
)
if self.mode in (SlackMode.WORKFLOW, SlackMode.WORKFLOW_TRIGGER):
return "{schema}://{path}/?{params}".format(
schema=self.secure_protocol,
path="/".join(
self.pprint(s, privacy, safe="")
for s in self.workflow_path
),
params=NotifySlack.urlencode(params),
)
if self.mode in (SlackMode.WEBHOOK, SlackMode.WEBHOOK_GOV):
return (
"{schema}://{botname}{token_a}/{token_b}/{token_c}/"
"{targets}/?{params}".format(
schema=self.secure_protocol,
botname=botname,
token_a=self.pprint(self.token_a, privacy, safe=""),
token_b=self.pprint(self.token_b, privacy, safe=""),
token_c=self.pprint(self.token_c, privacy, safe=""),
targets="/".join(
[NotifySlack.quote(x, safe="") for x in self.channels]
),
params=NotifySlack.urlencode(params),
)
)
# else -> self.mode == SlackMode.BOT:
return "{schema}://{botname}{access_token}/{targets}/?{params}".format(
schema=self.secure_protocol,
botname=botname,
access_token=self.pprint(self.access_token, privacy, safe=""),
targets="/".join(
[NotifySlack.quote(x, safe="") for x in self.channels]
),
params=NotifySlack.urlencode(params),
)
def __len__(self):
"""Returns the number of targets associated with this notification."""
return len(self.channels)
@staticmethod
def parse_url(url):
"""Parses the URL and returns enough arguments that can allow us to re-
instantiate this object."""
results = NotifyBase.parse_url(url, verify_host=False)
if not results:
# We're done early as we couldn't load the results
return results
# The first token is stored in the hostname
token = NotifySlack.unquote(results["host"])
# Get unquoted path entries
entries = NotifySlack.split_path(results["fullpath"])
# Peek at mode early to guide path parsing.
# Resolve via the same prefix-matching used in __init__ so that
# abbreviated inputs (e.g. mode=tri or mode=work) are handled
# identically in both places.
_mode_raw = results["qsd"].get("mode", "")
_mode = (
next(
(a for a in SLACK_MODES if a.startswith(_mode_raw)),
"",
)
if _mode_raw
else ""
)
if _mode in (SlackMode.WORKFLOW, SlackMode.WORKFLOW_TRIGGER):
# All path segments form the workflow_path
results["workflow_path"] = "/".join([token, *entries])
results["targets"] = []
# Verify if our token is a bot token or part of a webhook:
elif token.startswith("xo"):
# We're dealing with a bot
results["access_token"] = token
results["targets"] = entries
else:
# We're dealing with a webhook
results["token_a"] = token
results["token_b"] = entries.pop(0) if entries else None
results["token_c"] = entries.pop(0) if entries else None
# assign remaining entries to the channels we wish to notify
results["targets"] = entries
# Support the token flag where you can set it to the bot token
# or the webhook token (with slash delimiters)
if "token" in results["qsd"] and len(results["qsd"]["token"]):
# Break our entries up into a list; we can ue the Channel
# list delimiter above since it doesn't contain any characters
# we don't otherwise accept anyway in our token
entries = list(
filter(
bool,
CHANNEL_LIST_DELIM.split(
NotifySlack.unquote(results["qsd"]["token"])
),
)
)
# check to see if we're dealing with a bot/user token
if entries and entries[0].startswith("xo"):
# We're dealing with a bot
results["access_token"] = entries[0]
results["token_a"] = None
results["token_b"] = None
results["token_c"] = None
else: # Webhook
results["access_token"] = None
results["token_a"] = entries.pop(0) if entries else None
results["token_b"] = entries.pop(0) if entries else None
results["token_c"] = entries.pop(0) if entries else None
# Support the 'to' variable so that we can support rooms this way too
# The 'to' makes it easier to use yaml configuration
if "to" in results["qsd"] and len(results["qsd"]["to"]):
results["targets"] += list(
filter(
bool,
CHANNEL_LIST_DELIM.split(
NotifySlack.unquote(results["qsd"]["to"])
),
)
)
# Get Image Flag
results["include_image"] = parse_bool(
results["qsd"].get(
"image", NotifySlack.template_args["image"]["default"]
)
)
results["include_timestamp"] = parse_bool(
results["qsd"].get(
"timestamp", NotifySlack.template_args["timestamp"]["default"]
)
)
# Get Payload structure (use blocks?)
if "blocks" in results["qsd"] and len(results["qsd"]["blocks"]):
results["use_blocks"] = parse_bool(results["qsd"]["blocks"])
# Get Footer Flag
results["include_footer"] = parse_bool(
results["qsd"].get(
"footer", NotifySlack.template_args["footer"]["default"]
)
)
# Get Mode
if "mode" in results["qsd"] and len(results["qsd"]["mode"]):
results["mode"] = NotifySlack.unquote(results["qsd"]["mode"])
# Template Handling
if "template" in results["qsd"] and results["qsd"]["template"]:
results["template"] = NotifySlack.unquote(
results["qsd"]["template"]
)
# Store our tokens
results["tokens"] = results["qsd:"]
return results
@staticmethod
def parse_native_url(url):
"""
Supports:
- https://hooks.slack.com/services/TOKEN_A/TOKEN_B/TOKEN_C
- https://hooks.slack-gov.com/services/TOKEN_A/TOKEN_B/TOKEN_C
- https://hooks.slack.com/workflows/T.../F.../X.../Y...
- https://hooks.slack.com/triggers/T.../X.../Y...
"""
# Workflow Builder and trigger webhook URLs
wf_result = re.match(
r"^https?://hooks\.slack\.com/"
r"(?P<wf_type>workflows|triggers)/"
r"(?P<path>[A-Z0-9][A-Z0-9/_-]+)/?"
r"(?P<params>\?.+)?$",
url,
re.I,
)
if wf_result:
wf_type = wf_result.group("wf_type").lower()
# Map URL path type to Apprise mode string
mode = (
SlackMode.WORKFLOW_TRIGGER
if wf_type == "triggers"
else SlackMode.WORKFLOW
)
path = wf_result.group("path").strip("/")
params = wf_result.group("params") or ""
sep = "&" if params else "?"
return NotifySlack.parse_url(
"{schema}://{path}/{params}{sep}mode={mode}".format(
schema=NotifySlack.secure_protocol,
path=path,
params=params,
sep=sep,
mode=mode,
)
)
# Standard incoming webhook and government webhook URLs
result = re.match(
r"^https?://(?P<host>hooks\.slack(?P<gov>-gov)?\.com)/services/"
r"(?P<token_a>[A-Z0-9]+)/"
r"(?P<token_b>[A-Z0-9]+)/"
r"(?P<token_c>[A-Z0-9]+)/?"
r"(?P<params>\?.+)?$",
url,
re.I,
)
if result:
params = (
"" if not result.group("params") else result.group("params")
)
if result.group("gov"):
# provide gov parameters
params = (
"?" if not params else "&"
) + f"mode={SlackMode.WEBHOOK_GOV}"
return NotifySlack.parse_url(
"{schema}://{token_a}/{token_b}/{token_c}/{params}".format(
schema=NotifySlack.secure_protocol,
token_a=result.group("token_a"),
token_b=result.group("token_b"),
token_c=result.group("token_c"),
params=params,
)
)
return None
Reference in New Issue View Git Blame Copy Permalink