Added Resend Support (#1311)

This commit is contained in:
Chris Caron 2025-03-30 12:48:45 -04:00 committed by GitHub
parent efd9d5cd67
commit 689fc07e65
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 661 additions and 3 deletions

View File

@ -78,6 +78,7 @@ PushSafer
Pushy
PushDeer
Reddit
Resend
Revolt
Rocket.Chat
RSyslog

View File

@ -113,6 +113,7 @@ The table below identifies the services this tool supports and some example serv
| [Pushy](https://github.com/caronc/apprise/wiki/Notify_pushy) | pushy:// | (TCP) 443 | pushy://apikey/DEVICE<br />pushy://apikey/DEVICE1/DEVICE2/DEVICEN<br />pushy://apikey/TOPIC<br />pushy://apikey/TOPIC1/TOPIC2/TOPICN
| [PushDeer](https://github.com/caronc/apprise/wiki/Notify_pushdeer) | pushdeer:// or pushdeers:// | (TCP) 80 or 443 | pushdeer://pushKey<br />pushdeer://hostname/pushKey<br />pushdeer://hostname:port/pushKey
| [Reddit](https://github.com/caronc/apprise/wiki/Notify_reddit) | reddit:// | (TCP) 443 | reddit://user:password@app_id/app_secret/subreddit<br />reddit://user:password@app_id/app_secret/sub1/sub2/subN
| [Resend](https://github.com/caronc/apprise/wiki/Notify_resend) | resend:// | (TCP) 443 | resend://APIToken:FromEmail/<br />resend://APIToken:FromEmail/ToEmail<br />resend://APIToken:FromEmail/ToEmail1/ToEmail2/ToEmailN/
| [Revolt](https://github.com/caronc/apprise/wiki/Notify_Revolt) | revolt:// | (TCP) 443 | revolt://bottoken/ChannelID<br />revolt://bottoken/ChannelID1/ChannelID2/ChannelIDN |
| [Rocket.Chat](https://github.com/caronc/apprise/wiki/Notify_rocketchat) | rocket:// or rockets:// | (TCP) 80 or 443 | rocket://user:password@hostname/RoomID/Channel<br />rockets://user:password@hostname:443/#Channel1/#Channel1/RoomID<br />rocket://user:password@hostname/#Channel<br />rocket://webhook@hostname<br />rockets://webhook@hostname/@User/#Channel
| [RSyslog](https://github.com/caronc/apprise/wiki/Notify_rsyslog) | rsyslog:// | (UDP) 514 | rsyslog://hostname<br />rsyslog://hostname/Facility

460
apprise/plugins/resend.py Normal file
View File

@ -0,0 +1,460 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2025, 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.
# You will need an API Key for this plugin to work.
# From the Settings -> API Keys you can click "Create API Key" if you don't
# have one already. The key must have at least the "Mail Send" permission
# to work.
#
# The schema to use the plugin looks like this:
# {schema}://{apikey}:{from_email}
#
# Your {from_email} must be comprissed of your Resend Authenticated
# Domain.
# Simple API Reference:
# - https://resend.com/onboarding
import requests
from json import dumps
from .base import NotifyBase
from .. import exception
from ..common import NotifyFormat
from ..common import NotifyType
from ..utils.parse import parse_list, is_email, validate_regex
from ..locale import gettext_lazy as _
RESEND_HTTP_ERROR_MAP = {
200: "Successful request.",
400: "Check that the parameters were correct.",
401: "The API key used was missing.",
403: "The API key used was invalid.",
404: "The resource was not found.",
429: "The rate limit was exceeded.",
}
class NotifyResend(NotifyBase):
"""
A wrapper for Notify Resend Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Resend'
# The services URL
service_url = 'https://resend.com'
# The default secure protocol
secure_protocol = 'resend'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_resend'
# Default to markdown
notify_format = NotifyFormat.HTML
# The default Email API URL to use
notify_url = 'https://api.resend.com/emails'
# Support attachments
attachment_support = True
# Allow 300 requests per minute.
# 60/300 = 0.2
request_rate_per_sec = 0.2
# The default subject to use if one isn't specified.
default_empty_subject = '<no subject>'
# Define object templates
templates = (
'{schema}://{apikey}:{from_email}',
'{schema}://{apikey}:{from_email}/{targets}',
)
# Define our template arguments
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'^[A-Z0-9._-]+$', 'i'),
},
'from_email': {
'name': _('Source Email'),
'type': 'string',
'required': True,
},
'target_email': {
'name': _('Target Email'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
'cc': {
'name': _('Carbon Copy'),
'type': 'list:string',
},
'bcc': {
'name': _('Blind Carbon Copy'),
'type': 'list:string',
},
})
def __init__(self, apikey, from_email, targets=None, cc=None,
bcc=None, **kwargs):
"""
Initialize Notify Resend Object
"""
super().__init__(**kwargs)
# API Key (associated with project)
self.apikey = validate_regex(
apikey, *self.template_tokens['apikey']['regex'])
if not self.apikey:
msg = 'An invalid Resend API Key ' \
'({}) was specified.'.format(apikey)
self.logger.warning(msg)
raise TypeError(msg)
result = is_email(from_email)
if not result:
msg = 'Invalid ~From~ email specified: {}'.format(from_email)
self.logger.warning(msg)
raise TypeError(msg)
# Store email address
self.from_email = result['full_email']
# Acquire Targets (To Emails)
self.targets = list()
# Acquire Carbon Copies
self.cc = set()
# Acquire Blind Carbon Copies
self.bcc = set()
# Validate recipients (to:) and drop bad ones:
for recipient in parse_list(targets):
result = is_email(recipient)
if result:
self.targets.append(result['full_email'])
continue
self.logger.warning(
'Dropped invalid email '
'({}) specified.'.format(recipient),
)
# Validate recipients (cc:) and drop bad ones:
for recipient in parse_list(cc):
result = is_email(recipient)
if result:
self.cc.add(result['full_email'])
continue
self.logger.warning(
'Dropped invalid Carbon Copy email '
'({}) specified.'.format(recipient),
)
# Validate recipients (bcc:) and drop bad ones:
for recipient in parse_list(bcc):
result = is_email(recipient)
if result:
self.bcc.add(result['full_email'])
continue
self.logger.warning(
'Dropped invalid Blind Carbon Copy email '
'({}) specified.'.format(recipient),
)
if len(self.targets) == 0:
# Notify ourselves
self.targets.append(self.from_email)
return
@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.
"""
return (self.secure_protocol, self.apikey, self.from_email)
def url(self, privacy=False, *args, **kwargs):
"""
Returns the URL built dynamically based on specified arguments.
"""
# Our URL parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
if len(self.cc) > 0:
# Handle our Carbon Copy Addresses
params['cc'] = ','.join(self.cc)
if len(self.bcc) > 0:
# Handle our Blind Carbon Copy Addresses
params['bcc'] = ','.join(self.bcc)
# a simple boolean check as to whether we display our target emails
# or not
has_targets = \
not (len(self.targets) == 1 and self.targets[0] == self.from_email)
return '{schema}://{apikey}:{from_email}/{targets}?{params}'.format(
schema=self.secure_protocol,
apikey=self.pprint(self.apikey, privacy, safe=''),
# never encode email since it plays a huge role in our hostname
from_email=self.from_email,
targets='' if not has_targets else '/'.join(
[NotifyResend.quote(x, safe='') for x in self.targets]),
params=NotifyResend.urlencode(params),
)
def __len__(self):
"""
Returns the number of targets associated with this notification
"""
return len(self.targets)
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
**kwargs):
"""
Perform Resend Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/json',
'Authorization': 'Bearer {}'.format(self.apikey),
}
# error tracking (used for function return)
has_error = False
_payload = {
'from': self.from_email,
# A subject is a requirement, so if none is specified we must
# set a default with at least 1 character or Resend will deny
# our request
'subject': title if title else self.default_empty_subject,
'text' if self.notify_format == NotifyFormat.TEXT
else 'html': body,
}
if attach and self.attachment_support:
attachments = []
# Send our attachments
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 Resend attachment {}.'.format(
attachment.url(privacy=True)))
return False
try:
attachments.append({
"content": attachment.base64(),
"filename": attachment.name
if attachment.name else f'file{no:03}.dat',
"type": "application/octet-stream",
"disposition": "attachment"
})
except exception.AppriseException:
# We could not access the attachment
self.logger.error(
'Could not access Resend attachment {}.'.format(
attachment.url(privacy=True)))
return False
self.logger.debug(
'Appending Resend attachment {}'.format(
attachment.url(privacy=True)))
# Append our attachments to the payload
_payload.update({
'attachments': attachments,
})
targets = list(self.targets)
while len(targets) > 0:
target = targets.pop(0)
# Create a copy of our template
payload = _payload.copy()
# unique cc/bcc list management
cc = (self.cc - self.bcc - set([target]))
bcc = (self.bcc - set([target]))
# Set our target
payload['to'] = target
if len(cc):
payload['cc'] = list(cc)
if len(bcc):
payload['bcc'] = list(bcc)
self.logger.debug('Resend POST URL: %s (cert_verify=%r)' % (
self.notify_url, self.verify_certificate,
))
self.logger.debug('Resend Payload: %s', str(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
self.notify_url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code not in (
requests.codes.ok, requests.codes.accepted):
# We had a problem
status_str = \
NotifyResend.http_response_code_lookup(
r.status_code, RESEND_HTTP_ERROR_MAP)
self.logger.warning(
'Failed to send Resend notification to {}: '
'{}{}error={}.'.format(
target,
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
# Mark our failure
has_error = True
continue
else:
self.logger.info(
'Sent Resend notification to {}.'.format(target))
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending Resend '
'notification to {}.'.format(target))
self.logger.debug('Socket Exception: %s' % str(e))
# Mark our failure
has_error = True
continue
return not has_error
@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)
if not results:
# We're done early as we couldn't load the results
return results
# Our URL looks like this:
# {schema}://{apikey}:{from_email}/{targets}
#
# which actually equates to:
# {schema}://{user}:{password}@{host}/{email1}/{email2}/etc..
# ^ ^ ^
# | | |
# apikey -from addr-
if not results.get('user'):
# An API Key as not properly specified
return None
if not results.get('password'):
# A From Email was not correctly specified
return None
# Prepare our API Key
results['apikey'] = NotifyResend.unquote(results['user'])
# Prepare our From Email Address
results['from_email'] = '{}@{}'.format(
NotifyResend.unquote(results['password']),
NotifyResend.unquote(results['host']),
)
# Acquire our targets
results['targets'] = NotifyResend.split_path(results['fullpath'])
# The 'to' makes it easier to use yaml configuration
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifyResend.parse_list(results['qsd']['to'])
# Handle Carbon Copy Addresses
if 'cc' in results['qsd'] and len(results['qsd']['cc']):
results['cc'] = \
NotifyResend.parse_list(results['qsd']['cc'])
# Handle Blind Carbon Copy Addresses
if 'bcc' in results['qsd'] and len(results['qsd']['bcc']):
results['bcc'] = \
NotifyResend.parse_list(results['qsd']['bcc'])
return results

View File

@ -48,9 +48,9 @@ Windows, Microsoft Teams, Misskey, MQTT, MSG91, MyAndroid, Nexmo, Nextcloud,
NextcloudTalk, Notica, Notifiarr, Notifico, ntfy, Office365, OneSignal,
Opsgenie, PagerDuty, PagerTree, ParsePlatform, Plivo, PopcornNotify, Prowl,
Pushalot, PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, PushDeer,
Revolt, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Seven, SFR, Signal,
SimplePush, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost, Splunk,
Super Toasty, Streamlabs, Stride, Synology Chat, Syslog, Techulus Push,
Revolt, Reddit, Resend, Rocket.Chat, RSyslog, SendGrid, ServerChan, Seven, SFR,
Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost,
Splunk, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog, Techulus Push,
Telegram, Threema Gateway, Twilio, Twitter, Twist, VictorOps, Voipms, Vonage,
WeCom Bot, WhatsApp, Webex Teams, Workflows, WxPusher, XBMC}

196
test/test_plugin_resend.py Normal file
View File

@ -0,0 +1,196 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2025, 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.
from unittest import mock
import os
import pytest
import requests
from apprise import Apprise
from apprise import NotifyType
from apprise import AppriseAttachment
from apprise.plugins.resend import NotifyResend
from helpers import AppriseURLTester
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Attachment Directory
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
# a resend api key
UUID4 = 're_FmABCDEF_987654321zqbabc123abc8fw'
# Our Testing URLs
apprise_url_tests = (
('resend://', {
'instance': None,
}),
('resend://:@/', {
'instance': None,
}),
('resend://abcd', {
# Just an broken email (no api key or email)
'instance': None,
}),
('resend://abcd@host', {
# Just an Email specified, no API Key
'instance': None,
}),
('resend://invalid-api-key+*-d:user@example.com', {
# An invalid API Key
'instance': TypeError,
}),
('resend://abcd:user@example.com', {
# No To/Target Address(es) specified; so we sub in the same From
# address
'instance': NotifyResend,
}),
('resend://abcd:user@example.com/newuser@example.com', {
# A good email
'instance': NotifyResend,
}),
('resend://abcd:user@example.com/newuser@example.com'
'?bcc=l2g@nuxref.com', {
# A good email with Blind Carbon Copy
'instance': NotifyResend,
}),
('resend://abcd:user@example.com/newuser@example.com'
'?cc=l2g@nuxref.com', {
# A good email with Carbon Copy
'instance': NotifyResend,
}),
('resend://abcd:user@example.com/newuser@example.com'
'?to=l2g@nuxref.com', {
# A good email with Carbon Copy
'instance': NotifyResend,
}),
('resend://abcd:user@example.ca/newuser@example.ca', {
'instance': NotifyResend,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('resend://abcd:user@example.uk/newuser@example.uk', {
'instance': NotifyResend,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('resend://abcd:user@example.au/newuser@example.au', {
'instance': NotifyResend,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
)
def test_plugin_resend_urls():
"""
NotifyResend() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_plugin_resend_edge_cases(mock_post, mock_get):
"""
NotifyResend() Edge Cases
"""
# no apikey
with pytest.raises(TypeError):
NotifyResend(
apikey=None, from_email='user@example.com')
# invalid from email
with pytest.raises(TypeError):
NotifyResend(
apikey='abcd', from_email='!invalid')
# no email
with pytest.raises(TypeError):
NotifyResend(apikey='abcd', from_email=None)
# Invalid To email address
NotifyResend(
apikey='abcd', from_email='user@example.com', targets="!invalid")
# Test invalid bcc/cc entries mixed with good ones
assert isinstance(NotifyResend(
apikey='abcd',
from_email='l2g@example.com',
bcc=('abc@def.com', '!invalid'),
cc=('abc@test.org', '!invalid')), NotifyResend)
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_plugin_resend_attachments(mock_post, mock_get):
"""
NotifyResend() Attachments
"""
request = mock.Mock()
request.status_code = requests.codes.ok
# Prepare Mock
mock_post.return_value = request
mock_get.return_value = request
path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
attach = AppriseAttachment(path)
obj = Apprise.instantiate('resend://abcd:user@example.com')
assert isinstance(obj, NotifyResend)
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is True
mock_post.reset_mock()
mock_get.reset_mock()
# Try again in a use case where we can't access the file
with mock.patch("os.path.isfile", return_value=False):
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is False
# Try again in a use case where we can't access the file
with mock.patch("builtins.open", side_effect=OSError):
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
attach=attach) is False