Added support for 46elks:// (#1438)

This commit is contained in:
Chris Caron
2025-11-05 23:18:20 -05:00
committed by GitHub
parent eea03d1b68
commit 1ba14a7698
6 changed files with 528 additions and 8 deletions

View File

@@ -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<br/>46elks://user:password@FromPhoneNo/ToPhoneNo<br/>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<br/>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<br/>aprs://user:pass@callsign1/callsign2/callsignN
| [AWS SNS](https://github.com/caronc/apprise/wiki/Notify_sns) | sns:// | (TCP) 443 | sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo<br/>sns://AccessKeyID/AccessSecretKey/RegionName/+PhoneNo1/+PhoneNo2/+PhoneNoN<br/>sns://AccessKeyID/AccessSecretKey/RegionName/Topic<br/>sns://AccessKeyID/AccessSecretKey/RegionName/Topic1/Topic2/TopicN

View File

@@ -0,0 +1,369 @@
# 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.
"""
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<credentials>[^@]+)@"
r"api\.46elks\.com/a1/sms/?"
r"(?P<params>\?.+)$",
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

View File

@@ -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`, \

View File

@@ -51,6 +51,7 @@ dependencies = [
# Identifies all of the supported plugins
keywords = [
"46elks",
"Africas Talking",
"Alerts",
"Apprise API",

View File

@@ -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

View File

@@ -0,0 +1,142 @@
# 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.
# 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}")