remote watchtower: enforce that SSL is used, on the client-side

electrum/lnworker.py

@@ -17,6 +17,7 @@ from functools import partial
 from collections import defaultdict
 import concurrent
 from concurrent import futures
+import urllib.parse
 
 import dns.resolver
 import dns.exception
@@ -36,7 +37,7 @@ from .bip32 import BIP32Node
 from .util import bh2u, bfh, InvoiceError, resolve_dns_srv, is_ip_address, log_exceptions
 from .util import ignore_exceptions, make_aiohttp_session, SilentTaskGroup
 from .util import timestamp_to_datetime, random_shuffled_copy
-from .util import MyEncoder
+from .util import MyEncoder, is_private_netaddress
 from .logging import Logger
 from .lntransport import LNTransport, LNResponderTransport
 from .lnpeer import Peer, LN_P2P_NETWORK_TIMEOUT
@@ -531,10 +532,17 @@ class LNWallet(LNWorker):
     @log_exceptions
     async def sync_with_remote_watchtower(self):
         while True:
+            # periodically poll if the user updated 'watchtower_url'
             await asyncio.sleep(5)
             watchtower_url = self.config.get('watchtower_url')
             if not watchtower_url:
                 continue
+            parsed_url = urllib.parse.urlparse(watchtower_url)
+            if not (parsed_url.scheme == 'https' or is_private_netaddress(parsed_url.hostname)):
+                self.logger.warning(f"got watchtower URL for remote tower but we won't use it! "
+                                    f"can only use HTTPS (except if private IP): not using {watchtower_url!r}")
+                continue
+            # try to sync with the remote watchtower
             try:
                 async with make_aiohttp_session(proxy=self.network.proxy) as session:
                     watchtower = JsonRPCClient(session, watchtower_url)

electrum/tests/test_util.py

@@ -2,7 +2,7 @@ from decimal import Decimal
 
 from electrum.util import (format_satoshis, format_fee_satoshis, parse_URI,
                            is_hash256_str, chunks, is_ip_address, list_enabled_bits,
-                           format_satoshis_plain)
+                           format_satoshis_plain, is_private_netaddress)
 
 from . import ElectrumTestCase
 
@@ -148,3 +148,16 @@ class TestUtil(ElectrumTestCase):
         self.assertFalse(is_ip_address("2001:db8:0:0:g:ff00:42:8329"))
         self.assertFalse(is_ip_address("lol"))
         self.assertFalse(is_ip_address(":@ASD:@AS\x77\x22\xff¬!"))
+
+    def test_is_private_netaddress(self):
+        self.assertTrue(is_private_netaddress("127.0.0.1"))
+        self.assertTrue(is_private_netaddress("127.5.6.7"))
+        self.assertTrue(is_private_netaddress("::1"))
+        self.assertTrue(is_private_netaddress("[::1]"))
+        self.assertTrue(is_private_netaddress("localhost"))
+        self.assertTrue(is_private_netaddress("localhost."))
+        self.assertFalse(is_private_netaddress("[::2]"))
+        self.assertFalse(is_private_netaddress("2a00:1450:400e:80d::200e"))
+        self.assertFalse(is_private_netaddress("[2a00:1450:400e:80d::200e]"))
+        self.assertFalse(is_private_netaddress("8.8.8.8"))
+        self.assertFalse(is_private_netaddress("example.com"))

electrum/util.py

@@ -42,6 +42,7 @@ import time
 from typing import NamedTuple, Optional
 import ssl
 import ipaddress
+from ipaddress import IPv4Address, IPv6Address
 import random
 import attr
 
@@ -1238,6 +1239,19 @@ def is_ip_address(x: Union[str, bytes]) -> bool:
         return False
 
 
+def is_private_netaddress(host: str) -> bool:
+    if str(host) in ('localhost', 'localhost.',):
+        return True
+    if host[0] == '[' and host[-1] == ']':  # IPv6
+        host = host[1:-1]
+    try:
+        ip_addr = ipaddress.ip_address(host)  # type: Union[IPv4Address, IPv6Address]
+        return ip_addr.is_private
+    except ValueError:
+        pass  # not an IP
+    return False
+
+
 def list_enabled_bits(x: int) -> Sequence[int]:
     """e.g. 77 (0b1001101) --> (0, 2, 3, 6)"""
     binary = bin(x)[2:]