diff --git a/README.md b/README.md index 619866b6..aa878cc0 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,7 @@ SMS Notifications for the most part do not have a both a `title` and `body`. Th | Notification Service | Service ID | Default Port | Example Syntax | | -------------------- | ---------- | ------------ | -------------- | +| [46elks](https://github.com/caronc/apprise/wiki/Notify_46elks) | 46elks:// | (TCP) 443 | 46elks://user:password@FromPhoneNo
46elks://user:password@FromPhoneNo/ToPhoneNo
46elks://user:password@FromPhoneNo/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Africas Talking](https://github.com/caronc/apprise/wiki/Notify_africas_talking) | atalk:// | (TCP) 443 | atalk://AppUser@ApiKey/ToPhoneNo
atalk://AppUser@ApiKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/ | [Automated Packet Reporting System (ARPS)](https://github.com/caronc/apprise/wiki/Notify_aprs) | aprs:// | (TCP) 10152 | aprs://user:pass@callsign
aprs://user:pass@callsign1/callsign2/callsignN | [AWS SNS](https://github.com/caronc/apprise/wiki/Notify_sns) | sns:// | (TCP) 443 | sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo
sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo1/+PhoneNo2/+PhoneNoN
sns://AccessKeyID/AccessSecretKey/RegionName/Topic
sns://AccessKeyID/AccessSecretKey/RegionName/Topic1/Topic2/TopicN diff --git a/apprise/plugins/fortysixelks.py b/apprise/plugins/fortysixelks.py new file mode 100644 index 00000000..ad6193c4 --- /dev/null +++ b/apprise/plugins/fortysixelks.py @@ -0,0 +1,369 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, Chris Caron +# +# 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. +""" +46elks SMS Notification Service. + +Minimal URL formats (source ends up being target): + - 46elks://user:pass@/+15551234567 + - 46elks://user:pass@/+15551234567/+46701234567 + - 46elks://user:pass@/+15551234567?from=Acme +""" + +from __future__ import annotations + +from collections.abc import Iterable +import re +from typing import Any, Optional + +import requests + +from ..common import NotifyType +from ..locale import gettext_lazy as _ +from ..url import PrivacyMode +from ..utils.parse import ( + is_phone_no, + parse_phone_no, +) +from .base import NotifyBase + + +class Notify46Elks(NotifyBase): + """A wrapper for 46elks Notifications.""" + + # The default descriptive name associated with the Notification + service_name = _("46elks") + + # The services URL + service_url = "https://46elks.com" + + # The default secure protocol + secure_protocol = ("46elks", "elks") + + # A URL that takes you to the setup/help of the specific protocol + setup_url = "https://github.com/caronc/apprise/wiki/Notify_46elks" + + # 46elksAPI Request URLs + notify_url = "https://api.46elks.com/a1/sms" + + # The maximum allowable characters allowed in the title per message + title_maxlen = 0 + + # The maximum allowable characters allowed in the body per message + body_maxlen = 160 + + # Define object templates + templates = ( + "{schema}://{user}:{password}@/{from_phone}", + "{schema}://{user}:{password}@/{from_phone}/{targets}", + ) + + # Define our template tokens + template_tokens = dict( + NotifyBase.template_tokens, + **{ + "user": { + "name": _("API Username"), + "type": "string", + "required": True, + }, + "password": { + "name": _("API Password"), + "type": "string", + "private": True, + "required": True, + }, + "from_phone": { + "name": _("From Phone No"), + "type": "string", + "required": True, + "map_to": "source", + }, + "target_phone": { + "name": _("Target Phone"), + "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", + }, + "from": { + "alias_of": "from_phone", + }, + }, + ) + + def __init__( + self, + targets: Optional[Iterable[str]] = None, + source: Optional[str] = None, + **kwargs: Any, + ) -> None: + """ + Initialise 46elks notifier. + + :param targets: Iterable of phone numbers. E.164 is recommended. + :param source: Optional source ID or E.164 number. + """ + super().__init__(**kwargs) + + # Prepare our source + self.source: Optional[str] = (source or "").strip() or None + + if not self.password: + msg = "No 46elks password was specified." + self.logger.warning(msg) + raise TypeError(msg) + + elif not self.user: + msg = "No 46elks user was specified." + self.logger.warning(msg) + raise TypeError(msg) + + # Parse our targets + self.targets = [] + + if not targets and is_phone_no(self.source): + targets = [self.source] + + for target in parse_phone_no(targets): + # Validate targets and drop bad ones: + result = is_phone_no(target) + if not result: + self.logger.warning( + f"Dropped invalid phone # ({target}) specified.", + ) + continue + + # store valid phone number + # Carry forward '+' if defined, otherwise do not... + self.targets.append( + ("+" + result["full"]) + if target.lstrip()[0] == "+" + else result["full"] + ) + + def send( + self, + body: str, + title: str = "", + notify_type: NotifyType = NotifyType.INFO, + **kwargs: Any, + ) -> bool: + """Perform 46elks Notification.""" + + if not self.targets: + # There is no one to email; we're done + self.logger.warning( + "There are no 46elks recipients to notify" + ) + return False + + headers = { + "User-Agent": self.app_id, + } + + # error tracking (used for function return) + has_error = False + + targets = list(self.targets) + while targets: + target = targets.pop(0) + + # Prepare our payload + payload = { + "to": target, + "from": self.source, + "message": body, + } + + self.logger.debug( + "46elks POST URL:" + f" {self.notify_url} (cert_verify={self.verify_certificate!r})" + ) + self.logger.debug(f"46elks Payload: {payload!s}") + + # Always call throttle before any remote server i/o is made + self.throttle() + try: + r = requests.post( + self.notify_url, + data=payload, + headers=headers, + auth=(self.user, self.password), + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + if r.status_code != requests.codes.ok: + # We had a problem + status_str = ( + Notify46Elks.http_response_code_lookup( + r.status_code + ) + ) + + self.logger.warning( + "Failed to send 46elks notification to {}: " + "{}{}error={}.".format( + target, + status_str, + ", " if status_str else "", + r.status_code, + ) + ) + + self.logger.debug(f"Response Details:\r\n{r.content}") + + # Mark our failure + has_error = True + continue + + else: + self.logger.info( + f"Sent 46elks notification to {target}." + ) + + except requests.RequestException as e: + self.logger.warning( + "A Connection error occurred sending 46elks" + f" notification to {target}." + ) + self.logger.debug(f"Socket Exception: {e!s}") + + # Mark our failure + has_error = True + continue + + return not has_error + + @property + def url_identifier(self): + """Returns all of the identifiers that make this URL unique from + another similar one. + + Targets or end points should never be identified here. + """ + return (self.secure_protocol[0], self.user, self.password, self.source) + + def url(self, privacy: bool = False, *args: Any, **kwargs: Any) -> str: + """Returns the URL built dynamically based on specified arguments.""" + + # Initialize our parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + # Apprise URL can be condensed and target can be eliminated if its + # our source phone no + targets = ( + [] if len(self.targets) == 1 and + self.source in self.targets else self.targets) + + return "{schema}://{user}:{pw}@{source}/{targets}?{params}".format( + schema=self.secure_protocol[0], + user=self.quote(self.user, safe=""), + source=self.source if self.source else "", + pw=self.pprint( + self.password, privacy, mode=PrivacyMode.Secret, safe=""), + targets="/".join( + [Notify46Elks.quote(x, safe="+") for x in targets] + ), + params=Notify46Elks.urlencode(params), + ) + + def __len__(self): + """Returns the number of targets associated with this notification.""" + targets = len(self.targets) + return targets if targets > 0 else 1 + + @staticmethod + def parse_native_url(url): + """ + Support https://user:pw@api.46elks.com/a1/sms?to=+15551234567&from=Acme + """ + + result = re.match( + r"^https?://(?P[^@]+)@" + r"api\.46elks\.com/a1/sms/?" + r"(?P\?.+)$", + url, + re.I, + ) + + if result: + return Notify46Elks.parse_url( + "{schema}://{credentials}@/{params}".format( + schema=Notify46Elks.secure_protocol[0], + credentials=result.group("credentials"), + params=result.group("params"), + ) + ) + + return None + + @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 + + # Prepare our targets + results["targets"] = [] + + # The 'from' makes it easier to use yaml configuration + if "from" in results["qsd"] and len(results["qsd"]["from"]): + results["source"] = Notify46Elks.unquote( + results["qsd"]["from"] + ) + + elif results["host"]: + results["source"] = Notify46Elks.unquote(results["host"]) + + # Store our remaining targets found on path + results["targets"].extend( + Notify46Elks.split_path(results["fullpath"]) + ) + + # Support the 'to' variable so that we can support targets this way too + # The 'to' makes it easier to use yaml configuration + if "to" in results["qsd"] and len(results["qsd"]["to"]): + results["targets"] += Notify46Elks.parse_phone_no( + results["qsd"]["to"] + ) + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index e337c0dd..673ed931 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -56,8 +56,8 @@ Apprise is a Python package that simplifies access to many popular \ notification services. It supports sending alerts to platforms such as: \ \ -`AfricasTalking`, `Apprise API`, `APRS`, `AWS SES`, `AWS SNS`, `Bark`, \ -`BlueSky`, `Burst SMS`, `BulkSMS`, `BulkVS`, `Chanify`, `Clickatell`, \ +`46elks`, `AfricasTalking`, `Apprise API`, `APRS`, `AWS SES`, `AWS SNS`, +`Bark`, `BlueSky`, `Burst SMS`, `BulkSMS`, `BulkVS`, `Chanify`, `Clickatell`, \ `ClickSend`, `DAPNET`, `DingTalk`, `Discord`, `E-Mail`, `Emby`, `FCM`, \ `Feishu`, `Flock`, `Free Mobile`, `Google Chat`, `Gotify`, `Growl`, \ `Guilded`, `Home Assistant`, `httpSMS`, `IFTTT`, `Join`, `Kavenegar`, `KODI`, \ diff --git a/pyproject.toml b/pyproject.toml index 78be0783..6a2e5d22 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,7 @@ dependencies = [ # Identifies all of the supported plugins keywords = [ + "46elks", "Africas Talking", "Alerts", "Apprise API", diff --git a/tests/helpers/rest.py b/tests/helpers/rest.py index daab1667..775f8d41 100644 --- a/tests/helpers/rest.py +++ b/tests/helpers/rest.py @@ -118,11 +118,17 @@ class AppriseURLTester: def run(self, url, meta, tmpdir, mock_request, mock_post, mock_get): """Run a specific test.""" + if meta is False: + # Prepare a default structure to make life easy + meta = { + "instance": TypeError, + } + # Our expected instance - instance = meta.get("instance", None) + instance = meta.get("instance") # Our expected server objects - _self = meta.get("self", None) + _self = meta.get("self") # Our expected privacy url # Don't set this if don't need to check it's value @@ -250,7 +256,7 @@ class AppriseURLTester: privacy_url ): raise AssertionError( - "Privacy URL:" + f"URL: {url} Privacy URL:" f" '{obj.url(privacy=True)[:len(privacy_url)]}' !=" f" expected '{privacy_url}'" ) @@ -272,21 +278,22 @@ class AppriseURLTester: # Our new object should produce the same url identifier elif obj.url_identifier != obj_cmp.url_identifier: raise AssertionError( - f"URL Identifier: '{obj_cmp.url_identifier}' != expected" + f"URL: {url} URL Identifier: " + f"'{obj_cmp.url_identifier}' != expected" f" '{obj.url_identifier}'" ) # Back our check up if obj.url_id() != obj_cmp.url_id(): raise AssertionError( - f"URL ID(): '{obj_cmp.url_id()}' != expected" + f"URL: {url} URL ID(): '{obj_cmp.url_id()}' != expected" f" '{obj.url_id()}'" ) # Verify there is no change from the old and the new if len(obj) != len(obj_cmp): raise AssertionError( - f"Target miscount {len(obj)} != {len(obj_cmp)}" + f"URL: {url} target miscount {len(obj)} != {len(obj_cmp)}" ) # Tidy our object diff --git a/tests/test_plugin_fortysixelks.py b/tests/test_plugin_fortysixelks.py new file mode 100644 index 00000000..b4c5d149 --- /dev/null +++ b/tests/test_plugin_fortysixelks.py @@ -0,0 +1,142 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, Chris Caron +# +# 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. + +# Disable logging for a cleaner testing output +import logging +from unittest import mock + +from helpers import AppriseURLTester +import requests + +from apprise import Apprise, NotifyType +from apprise.plugins.fortysixelks import Notify46Elks + +logging.disable(logging.CRITICAL) + +# Our Testing URLs +apprise_url_tests = ( + ("46elks://", False), + ("46elks://user@/", False), + ("46elks://:pass@/", False), + + ("46elks://user:pass@/", { + "instance": Notify46Elks, + # no target was specified + "notify_response": False, + }), + + ("46elks://user:pass@+15551234556", { + "instance": Notify46Elks, + }), + + ("46elks://user:pass@+15551234567/+46701234534?from=Acme", { + "instance": Notify46Elks, + }), + + # Support elks:// too! + ("elks://user:pass@+15551234123/", { + "instance": Notify46Elks, + }), + + # Privacy mode redacts password + ("46elks://user:pass@+15551234512", { + "privacy_url": "46elks://user:****@+15551234512", + "instance": Notify46Elks, + }), + + # invalid phone no + ("46elks://user:pass@Acme/234512", { + "instance": Notify46Elks, + "notify_response": False, + }), + # Native URL reversal + (("https://user1:pass@" + "api.46elks.com/a1/sms?to=+15551234511&from=Acme"), { + "instance": Notify46Elks, + "privacy_url": "46elks://user1:****@Acme/+15551234511", + }), + ("46elks://user:pass@+15551234567", + { + "instance": Notify46Elks, + # throw a bizarre code forcing us to fail to look it up + "response": False, + "requests_response_code": 999, + }), + ("46elks://user:pass@+15551234578", + { + "instance": Notify46Elks, + # Throws a series of i/o exceptions with this flag + # is set and tests that we gracefully handle them + "test_requests_exceptions": True, + }), +) + + +def test_plugin_46elks_urls(): + """NotifyTemplate() Apprise URLs.""" + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch("requests.post") +def test_plugin_46elks_edge_cases(mock_post): + """Notify46Elks() Edge Cases.""" + + user = "user1" + password = "pass123" + phone = "+15551234591" + + response = requests.Request() + response.status_code = requests.codes.ok + + # Prepare Mock + mock_post.return_value = response + + obj = Apprise.instantiate(f"46elks://{user}:{password}@{phone}") + assert ( + obj.notify(body="body", title="title", notify_type=NotifyType.INFO) + is True + ) + + # We know there is 1 (valid) targets + assert len(obj) == 1 + + # Test our call count + assert mock_post.call_count == 1 + + # Test + details = mock_post.call_args_list[0] + headers = details[1]["headers"] + assert headers["User-Agent"] == "Apprise" + payload = details[1]["data"] + assert payload["to"] == phone + assert payload["from"] == phone + assert payload["message"] == "title\r\nbody" + + # Verify our URL looks good + assert obj.url().startswith(f"46elks://{user}:{password}@{phone}")