Coverage for src/local_deep_research/security/notification_destination.py: 87%
222 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Classify the network destination represented by an Apprise URL."""
3from dataclasses import dataclass
4from enum import Enum
5import ipaddress
6import re
7from string import ascii_letters, digits, hexdigits
8from urllib.parse import SplitResult, unquote, urlsplit
10from urllib3.exceptions import LocationParseError
11from urllib3.util import parse_url
14class NotificationDestinationKind(str, Enum):
15 HTTP = "http"
16 AUTHORITY_PLUGIN = "authority_plugin"
17 VENDOR_PLUGIN = "vendor_plugin"
20@dataclass(frozen=True, slots=True)
21class NotificationDestination:
22 kind: NotificationDestinationKind
23 effective_host: str | None
26class NotificationDestinationError(ValueError):
27 """A notification URL whose destination cannot be classified safely."""
29 def __init__(self, reason: str) -> None:
30 super().__init__(reason)
31 self.reason = reason
33 def __str__(self) -> str:
34 return self.reason
37_UNRESERVED_HOST_ESCAPES = re.compile(r"%([0-9A-Fa-f]{2})")
38_UNRESERVED_HOST_CHARACTERS = frozenset(ascii_letters + digits + "-._~")
41def normalize_host_for_transport(host: str) -> str:
42 """Decode one pass of unreserved host escapes as Requests prepares URLs."""
44 def unescape(match: re.Match[str]) -> str:
45 decoded = chr(int(match.group(1), 16))
46 return (
47 decoded
48 if decoded in _UNRESERVED_HOST_CHARACTERS
49 else match.group(0)
50 )
52 return _UNRESERVED_HOST_ESCAPES.sub(unescape, host)
55def _required_host(parsed: SplitResult) -> str:
56 try:
57 host = parsed.hostname
58 except ValueError as exc:
59 raise NotificationDestinationError("invalid authority") from exc
60 if not host:
61 raise NotificationDestinationError("destination host is missing")
62 if ":" in host and "%" in host:
63 # An IPv6 zone identifier (``[fd00::1%25eth0]``) scopes a literal to
64 # an interface; no notification endpoint needs one, and a scoped
65 # address no longer equals the unscoped metadata literals.
66 raise NotificationDestinationError("IPv6 zone identifier in host")
67 # urlsplit leaves percent escapes in the hostname. Requests decodes
68 # unreserved escapes while preparing the outbound URL, but urllib3's
69 # parse_url changed its own decoding behavior between 2.7 and 2.8.
70 # Normalize once so policy resolves the same host under either version.
71 authority = f"[{host}]" if ":" in host else host
72 try:
73 network_host = parse_url(f"http://{authority}/").host
74 except LocationParseError as exc:
75 raise NotificationDestinationError("invalid authority") from exc
76 if not network_host: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 raise NotificationDestinationError("destination host is missing")
78 if network_host.startswith("[") and network_host.endswith("]"):
79 network_host = network_host[1:-1]
80 return normalize_host_for_transport(network_host).rstrip(".")
83def _plugin_host(parsed: SplitResult) -> str:
84 """Return the authority host of an Apprise plugin URL.
86 Several plugins (Signal, ntfy, Matrix, email) parse with
87 ``verify_host=False`` and fully percent-decode the host, so an escape
88 such as ``%2F`` or ``%40`` moves the real destination
89 (``evil.com%2F.local`` is sent to ``evil.com``). No legitimate plugin
90 host needs an escape, so any ``%`` is refused instead of guessing each
91 plugin's decoding.
92 """
93 try:
94 raw_host = parsed.hostname
95 except ValueError as exc:
96 raise NotificationDestinationError("invalid authority") from exc
97 if raw_host and "%" in raw_host:
98 raise NotificationDestinationError(
99 "percent-encoded plugin host is ambiguous"
100 )
101 return _required_host(parsed)
104def parse_apprise_query(url: str) -> dict[str, list[str]]:
105 """Parse query keys with the same normalization Apprise applies."""
106 _, separator, query = url.partition("?")
107 if not separator or not query:
108 return {}
110 parsed_query: dict[str, list[str]] = {}
111 pairs = [
112 part for section in query.split("&") for part in section.split(";")
113 ]
114 for pair in pairs:
115 raw_key, _, raw_value = pair.partition("=")
116 index = 0
117 while index < len(raw_key):
118 if raw_key[index] != "%":
119 index += 1
120 continue
121 if ( 121 ↛ 126line 121 didn't jump to line 126 because the condition on line 121 was never true
122 index + 2 >= len(raw_key)
123 or raw_key[index + 1] not in hexdigits
124 or raw_key[index + 2] not in hexdigits
125 ):
126 raise NotificationDestinationError(
127 "Malformed percent-encoding in notification parameter name"
128 )
129 index += 3
131 apprise_key = (
132 raw_key[:1] + raw_key[1:].replace("+", " ") if raw_key else ""
133 )
134 canonical_key = unquote(apprise_key).lower().strip()
135 parsed_query.setdefault(canonical_key, []).append(
136 unquote(raw_value).strip()
137 )
138 return parsed_query
141def _query_value(url: str, key: str) -> str | None:
142 values = parse_apprise_query(url).get(key)
143 if values is None:
144 return None
145 if len(values) != 1 or not values[0].strip(): 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 raise NotificationDestinationError(f"ambiguous {key} parameter")
147 return values[0].strip().lower()
150def _apprise_host(host: object) -> str:
151 """Normalize a host reported by Apprise for comparison with ours."""
152 if not isinstance(host, str) or not host: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 raise NotificationDestinationError("destination host is missing")
154 if host.startswith("[") and host.endswith("]"):
155 host = host[1:-1]
156 return host.rstrip(".").lower()
159def _confirm_apprise_ipv6_host(url: str, host: str) -> str:
160 """Refuse an IPv6 literal that Apprise would send somewhere else.
162 Apprise's generic URL parser rewrites compressed IPv6 literals in the
163 authority: on Apprise 1.13 ``json://[fd00::1]/x`` builds a plugin whose
164 host is ``[fd00::]`` and ``[::ffff:10.0.0.5]`` becomes ``[::ffff:10]``.
165 Policy would then vet one address while Apprise connects to another.
166 For an IPv6 literal the plugin is therefore built the way
167 ``Apprise.add`` builds it (``Apprise.instantiate``, no network I/O) and
168 its host must be the same address. A URL Apprise cannot build is
169 returned unchanged: ``Apprise.add`` rejects it and nothing is sent.
170 Hostnames and IPv4 literals are not rewritten by that parser and are
171 returned as they are.
173 The port is compared too: Apprise's port regex for a bracketed literal
174 is case-sensitive, so ``json://[FD00:0:0:0:0:0:0:1]:8080/x`` keeps the
175 host but silently drops ``:8080`` and would post to the default port.
176 """
177 try:
178 expected = ipaddress.ip_address(host)
179 except ValueError:
180 return host
181 if expected.version != 6:
182 return host
184 from apprise import Apprise, AppriseAsset
186 plugin = Apprise.instantiate(
187 url, asset=AppriseAsset(async_mode=False, http_redirects=False)
188 )
189 if plugin is None:
190 return host
191 try:
192 actual = ipaddress.ip_address(
193 _apprise_host(getattr(plugin, "host", None))
194 )
195 except ValueError as exc:
196 raise NotificationDestinationError(
197 "IPv6 host is rewritten by Apprise"
198 ) from exc
199 if actual != expected:
200 raise NotificationDestinationError("IPv6 host is rewritten by Apprise")
201 try:
202 expected_port = urlsplit(url).port
203 except ValueError as exc:
204 raise NotificationDestinationError("IPv6 port is malformed") from exc
205 actual_port = getattr(plugin, "port", None)
206 if isinstance(actual_port, str) and actual_port.isdigit(): 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 actual_port = int(actual_port)
208 if actual_port != expected_port:
209 raise NotificationDestinationError("IPv6 port is rewritten by Apprise")
210 return host
213def _apprise_ntfy_destination(url: str, host: str) -> NotificationDestination:
214 """Classify an ntfy URL from the plugin object Apprise itself builds.
216 ntfy picks cloud mode (a fixed POST to ntfy.sh, with the authority used
217 as a topic) or private mode (a POST to the authority host) from the
218 host's validity, the path, ``to=`` and ``mode=``. Re-implementing those
219 rules drifts in both directions, so the decision is read from the plugin
220 built the way ``Apprise.instantiate`` (called by ``Apprise.add``) builds
221 it: ``url_to_dict`` followed by the plugin constructor. Construction
222 performs no network I/O.
223 """
224 from apprise import AppriseAsset
225 from apprise.plugins import url_to_dict
226 from apprise.plugins.ntfy import NotifyNtfy
228 results = url_to_dict(url)
229 if not results or str(results.get("schema", "")).lower() not in (
230 "ntfy",
231 "ntfys",
232 ):
233 raise NotificationDestinationError(
234 "ntfy URL is not accepted by Apprise"
235 )
236 try:
237 plugin = NotifyNtfy(
238 **{
239 **results,
240 "asset": AppriseAsset(async_mode=False, http_redirects=False),
241 "tag": set(),
242 }
243 )
244 except Exception as exc: # Apprise.instantiate treats any error as reject
245 raise NotificationDestinationError(
246 "ntfy URL is not accepted by Apprise"
247 ) from exc
248 mode = str(getattr(plugin, "mode", "") or "").strip().lower()
249 if mode == "cloud":
250 return NotificationDestination(
251 NotificationDestinationKind.VENDOR_PLUGIN,
252 None,
253 )
254 if mode != "private": 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 raise NotificationDestinationError("unsupported ntfy mode")
256 if not getattr(plugin, "topics", None):
257 # Apprise would send nothing; refuse rather than guess.
258 raise NotificationDestinationError("private ntfy target is missing")
259 apprise_host = _apprise_host(getattr(plugin, "host", None))
260 if apprise_host != host: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 raise NotificationDestinationError("ntfy host is ambiguous")
262 if apprise_host == "ntfy.sh":
263 return NotificationDestination(
264 NotificationDestinationKind.VENDOR_PLUGIN,
265 None,
266 )
267 return NotificationDestination(
268 NotificationDestinationKind.AUTHORITY_PLUGIN,
269 apprise_host,
270 )
273def _apprise_matrix_details(url: str) -> tuple[bool, str]:
274 """Return Apprise's own (is_t2bot, host) for a Matrix URL.
276 Uses the parser ``Apprise.add`` calls (``url_to_dict``) without building
277 the plugin, whose constructor reads Apprise's persistent store.
278 """
279 from apprise.plugins import url_to_dict
281 results = url_to_dict(url)
282 if not results or str(results.get("schema", "")).lower() != "matrix": 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 raise NotificationDestinationError(
284 "matrix URL is not accepted by Apprise"
285 )
286 mode = str(results.get("mode") or "").strip().lower()
287 return mode == "t2bot", str(results.get("host") or "")
290def _email_template_smtp_hosts() -> frozenset[str]:
291 """Return the fixed SMTP hosts of Apprise's built-in mail providers."""
292 from apprise.plugins.email.templates import EMAIL_TEMPLATES
294 return frozenset(
295 str(template[2]["smtp_host"]).rstrip(".").lower()
296 for template in EMAIL_TEMPLATES
297 if template[2].get("smtp_host")
298 )
301def _apprise_mail_host(url: str, host: str) -> str:
302 """Confirm that Apprise's mail plugin connects to ``host`` or a provider.
304 Apprise's email parser discards an authority it does not accept as a
305 hostname (an empty port, an underscore) and then takes the SMTP host from
306 the domain of an e-mail address in the userinfo or ``?user=``. The URL
307 authority would then no longer be the host Apprise connects to. The
308 plugin is therefore built the way ``Apprise.instantiate`` builds it
309 (``url_to_dict`` and the constructor, no network I/O), and only two
310 outcomes are accepted: Apprise keeps ``host`` and connects to it, or it
311 connects to the fixed SMTP host of one of its built-in providers selected
312 by ``host``. A URL whose plugin cannot be built is returned unchanged:
313 ``Apprise.add`` builds the same plugin, rejects the URL and sends nothing.
314 """
315 from apprise import AppriseAsset
316 from apprise.plugins import url_to_dict
317 from apprise.plugins.email import NotifyEmail
319 results = url_to_dict(url)
320 if not results or str(results.get("schema", "")).lower() != "mailto": 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 raise NotificationDestinationError(
322 "mailto URL is not accepted by Apprise"
323 )
324 parsed_host = results.get("host")
325 if not parsed_host or _apprise_host(parsed_host) != host:
326 raise NotificationDestinationError("mail host is ambiguous")
327 try:
328 plugin = NotifyEmail(
329 **{
330 **results,
331 "asset": AppriseAsset(async_mode=False, http_redirects=False),
332 "tag": set(),
333 }
334 )
335 except Exception: # Apprise.instantiate treats any error as reject
336 return host
337 smtp_host = _apprise_host(getattr(plugin, "smtp_host", None))
338 if smtp_host != host and smtp_host not in _email_template_smtp_hosts(): 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 raise NotificationDestinationError("mail host is ambiguous")
340 return host
343def _smtp_override_host(value: str) -> str:
344 if any(character.isspace() for character in value): 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 raise NotificationDestinationError("invalid smtp override")
346 try:
347 parsed = urlsplit(f"//{value}")
348 port = parsed.port
349 except ValueError as exc:
350 raise NotificationDestinationError("invalid smtp override") from exc
351 if ( 351 ↛ 359line 351 didn't jump to line 359 because the condition on line 351 was never true
352 parsed.username is not None
353 or parsed.password is not None
354 or port is not None
355 or parsed.path
356 or parsed.query
357 or parsed.fragment
358 ):
359 raise NotificationDestinationError("invalid smtp override")
360 if "%" in value: 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 raise NotificationDestinationError("invalid smtp override")
362 return _required_host(parsed)
365def parse_notification_destination(url: str) -> NotificationDestination:
366 """Return the effective outbound destination for one Apprise URL."""
367 if not isinstance(url, str) or not url.strip():
368 raise NotificationDestinationError("notification URL is empty")
369 try:
370 normalized_url = url.strip()
371 parsed = urlsplit(normalized_url)
372 except ValueError as exc:
373 raise NotificationDestinationError(
374 "notification URL is malformed"
375 ) from exc
377 scheme = parsed.scheme.lower()
378 match scheme:
379 case "http" | "https":
380 return NotificationDestination(
381 NotificationDestinationKind.HTTP,
382 _confirm_apprise_ipv6_host(
383 normalized_url, _required_host(parsed)
384 ),
385 )
386 case (
387 "json"
388 | "xml"
389 | "form"
390 | "gotify"
391 | "signal"
392 | "mattermost"
393 | "rocketchat"
394 ):
395 return NotificationDestination(
396 NotificationDestinationKind.AUTHORITY_PLUGIN,
397 _confirm_apprise_ipv6_host(
398 normalized_url, _plugin_host(parsed)
399 ),
400 )
401 case "discord" | "slack" | "tgram" | "pushover" | "teams":
402 if not parsed.netloc: 402 ↛ 403line 402 didn't jump to line 403 because the condition on line 402 was never true
403 raise NotificationDestinationError("vendor token is missing")
404 return NotificationDestination(
405 NotificationDestinationKind.VENDOR_PLUGIN,
406 None,
407 )
408 case "ntfy" | "ntfys":
409 # A duplicate or empty mode is ambiguous; fail before Apprise
410 # silently picks one spelling.
411 mode = _query_value(normalized_url, "mode")
412 if mode not in (None, "cloud", "private"):
413 raise NotificationDestinationError("unsupported ntfy mode")
414 return _apprise_ntfy_destination(
415 normalized_url, _plugin_host(parsed)
416 )
417 case "mailto":
418 smtp_override = _query_value(normalized_url, "smtp")
419 effective_host = (
420 _smtp_override_host(smtp_override)
421 if smtp_override is not None
422 else _apprise_mail_host(normalized_url, _plugin_host(parsed))
423 )
424 return NotificationDestination(
425 NotificationDestinationKind.AUTHORITY_PLUGIN,
426 effective_host,
427 )
428 case "matrix":
429 mode = _query_value(normalized_url, "mode")
430 if mode not in (
431 None,
432 "off",
433 "matrix",
434 "slack",
435 "hookshot",
436 "t2bot",
437 ):
438 raise NotificationDestinationError("unsupported matrix mode")
439 # Apprise alone decides between t2bot (a fixed POST to
440 # webhooks.t2bot.io, with the authority used as the token) and a
441 # homeserver/webhook mode that sends to the authority host.
442 # Mirroring that rule here drifted from Apprise and refused
443 # documented forms such as ``user@<token>`` and ``/#room``.
444 apprise_t2bot, apprise_host = _apprise_matrix_details(
445 normalized_url
446 )
447 if apprise_t2bot:
448 token = parsed.netloc.rpartition("@")[2]
449 if not re.fullmatch(r"[A-Za-z0-9]{64}", token):
450 raise NotificationDestinationError(
451 "vendor token is invalid"
452 )
453 return NotificationDestination(
454 NotificationDestinationKind.VENDOR_PLUGIN,
455 None,
456 )
457 host = _plugin_host(parsed)
458 if _apprise_host(apprise_host) != host: 458 ↛ 459line 458 didn't jump to line 459 because the condition on line 458 was never true
459 raise NotificationDestinationError("matrix host is ambiguous")
460 return NotificationDestination(
461 NotificationDestinationKind.AUTHORITY_PLUGIN,
462 host,
463 )
464 case _:
465 raise NotificationDestinationError(
466 "unsupported notification scheme"
467 )