Pin LNURL connections to validated public addresses

docker/justworks/business/server.py

@@ -6,6 +6,8 @@ import mimetypes
 import os
 import secrets
 import socket
+import ssl
+import http.client
 import sqlite3
 import subprocess
 import threading
@@ -128,48 +130,66 @@ def screen_tips(npub, lightning_address):
     name, host = lightning_address.split('@')
     if not name or not host or any(char in host for char in '/?#:@'):
         raise ValueError('invalid Lightning address')
-    endpoint = assert_public_https(f'https://{host}/.well-known/lnurlp/{quote(name, safe="")}')
+    endpoint = f'https://{host}/.well-known/lnurlp/{quote(name, safe="")}'
-
+    pay = fetch_lnurl_json(endpoint, timeout=4)
-    class NoRedirect(HTTPRedirectHandler):
-        def redirect_request(self, *args, **kwargs):
-            raise ValueError('LNURL redirect is not permitted for screen metadata')
-
-    with build_opener(NoRedirect).open(Request(endpoint, headers={'User-Agent': 'JustWorks-Screen/1'}), timeout=4) as response:
-        raw = response.read(65537)
-        if len(raw) > 65536:
-            raise ValueError('LNURL metadata too large')
-        pay = json.loads(raw)
     provider = pay.get('nostrPubkey')
     if pay.get('tag') != 'payRequest' or pay.get('allowsNostr') is not True or not isinstance(provider, str) or len(provider) != 64 or any(c not in '0123456789abcdef' for c in provider):
         raise ValueError('No verified Nostr receipt source')
     return helper({'action': 'screen-tips', 'npub': npub, 'provider': provider, 'relays': PAYMENT_RELAYS})['tips']
 
 
-def assert_public_https(url):
+def resolve_public_https(url):
     parsed = urlparse(url)
-    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
+    if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
         raise ValueError("Lightning provider returned an unsafe URL")
     addresses = socket.getaddrinfo(parsed.hostname, parsed.port or 443, type=socket.SOCK_STREAM)
     if not addresses or any(not ipaddress.ip_address(item[4][0]).is_global for item in addresses):
         raise ValueError("Lightning provider must use a public host")
-    return url
+    return parsed, addresses
 
 
-class NoLnurlRedirect(HTTPRedirectHandler):
+class PinnedLnurlConnection(http.client.HTTPSConnection):
-    def redirect_request(self, *args, **kwargs):
+    def __init__(self, parsed, addresses, timeout):
-        # A validated public endpoint cannot delegate access to another host or
+        # Direct sockets intentionally ignore ambient HTTP(S)_PROXY settings.
-        # protocol, especially a private service reachable from this node.
+        super().__init__(parsed.hostname, parsed.port or 443, timeout=timeout,
-        raise ValueError("Lightning provider redirects are not permitted")
+                         context=ssl.create_default_context())
+        self.addresses = addresses
+
+    def connect(self):
+        last_error = None
+        for family, kind, protocol, _canonical, address in self.addresses:
+            connection = socket.socket(family, kind, protocol)
+            try:
+                connection.settimeout(self.timeout)
+                # Numeric sockaddr from the validated lookup: no second DNS lookup.
+                connection.connect(address)
+                self.sock = self._context.wrap_socket(connection, server_hostname=self.host)
+                return
+            except OSError as error:
+                connection.close()
+                last_error = error
+        raise last_error or OSError("Lightning provider connection unavailable")
 
 
-def fetch_lnurl_json(url):
+def fetch_lnurl_json(url, timeout=10):
-    endpoint = assert_public_https(url)
+    parsed, addresses = resolve_public_https(url)
-    request = Request(endpoint, headers={"User-Agent": "JustWorks-Business/0.1"})
+    connection = PinnedLnurlConnection(parsed, addresses, timeout)
-    with build_opener(NoLnurlRedirect).open(request, timeout=10) as response:
+    try:
+        target = parsed.path or "/"
+        if parsed.query:
+            target += "?" + parsed.query
+        connection.request("GET", target, headers={"User-Agent": "JustWorks-Business/0.1"})
+        response = connection.getresponse()
+        if 300 <= response.status < 400:
+            raise ValueError("Lightning provider redirects are not permitted")
+        if response.status >= 400:
+            raise HTTPError(url, response.status, response.reason, response.headers, None)
         raw = response.read(65537)
-    if len(raw) > 65536:
+        if len(raw) > 65536:
-        raise ValueError("Lightning provider response too large")
+            raise ValueError("Lightning provider response too large")
-    return json.loads(raw)
+        return json.loads(raw)
+    finally:
+        connection.close()
 
 
 def lightning_invoice(lightning_address, amount_msat, comment=""):
@@ -178,14 +198,14 @@ def lightning_invoice(lightning_address, amount_msat, comment=""):
     name, host = lightning_address.rsplit("@", 1)
     if not name or not host or any(char in host for char in "/?#"):
         raise ValueError("Merchant Lightning address is invalid")
-    endpoint = assert_public_https(f"https://{host}/.well-known/lnurlp/{name}")
+    endpoint = f"https://{host}/.well-known/lnurlp/{quote(name, safe="")}"
     pay = fetch_lnurl_json(endpoint)
     if pay.get("tag") != "payRequest" or not pay.get("callback"):
         raise ValueError("Lightning address does not support payments")
     minimum, maximum = int(pay.get("minSendable", 0)), int(pay.get("maxSendable", 0))
     if amount_msat < minimum or (maximum and amount_msat > maximum):
         raise ValueError(f"Amount must be between {max(1, minimum // 1000)} and {maximum // 1000} sats")
-    callback = assert_public_https(str(pay["callback"]))
+    callback = str(pay["callback"])
     # Some LNURL providers reject otherwise valid NIP-57 requests when their
     # signed content contains a literal percent sign. Keep the human meaning
     # while using a provider-safe comment for both the zap and callback.

docker/justworks/business/test_lnurl_transport.py

@@ -1,52 +1,70 @@
-"""No-network checks of the LNURL transport trust boundary."""
+"""No-network checks using real HTTP request/response parsing over fake sockets."""
 import io
+import socket
 import unittest
-from email.message import Message
+from unittest.mock import Mock, patch
-from unittest.mock import patch
 from urllib.error import HTTPError
-from urllib.request import HTTPSHandler, build_opener
-from urllib.response import addinfourl
 import server
 
-class FixtureHttps(HTTPSHandler):
+PUBLIC = [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('93.184.216.34', 443))]
+PRIVATE = [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('127.0.0.1', 443))]
+
+class FixtureSocket:
     def __init__(self, body=b'{}', redirect=None, status=302):
-        super().__init__()
+        headers = f'Location: {redirect}\r\n' if redirect else ''
-        self.body, self.redirect, self.status, self.calls = body, redirect, status, []
+        self.response = (f'HTTP/1.1 {status if redirect else 200} Fixture\r\n{headers}Content-Length: {len(body)}\r\nConnection: close\r\n\r\n').encode() + body
-    def https_open(self, request):
+        self.connected, self.sent, self.closed, self.timeout = [], b'', False, None
-        self.calls.append(request.full_url)
+    def settimeout(self, value): self.timeout = value
-        headers = Message()
+    def connect(self, address): self.connected.append(address)
-        if self.redirect: headers['Location'] = self.redirect
+    def sendall(self, body): self.sent += body
-        response = addinfourl(io.BytesIO(self.body), headers, request.full_url, self.status if self.redirect else 200)
+    def makefile(self, *args): return io.BytesIO(self.response)
-        response.msg = 'Found' if self.redirect else 'OK'
+    def close(self): self.closed = True
-        return response
 
 class LnurlTransportTests(unittest.TestCase):
-    def transport(self, fixture):
+    def transport(self, fixture, url='https://provider.example/lnurl', timeout=10):
-        with patch('server.assert_public_https', side_effect=lambda value: value), patch('server.build_opener', side_effect=lambda *handlers: build_opener(*handlers, fixture)):
+        context = Mock()
-            return server.fetch_lnurl_json('https://provider.example/lnurl')
+        context.wrap_socket.return_value = fixture
-    def test_actual_redirect_chain_refuses_internal_and_public_destinations(self):
+        with patch('server.socket.getaddrinfo', side_effect=[PUBLIC, PRIVATE]) as lookup, patch('server.socket.socket', return_value=fixture), patch('server.ssl.create_default_context', return_value=context), patch.dict('os.environ', {'HTTPS_PROXY':'http://127.0.0.1:9999'}):
+            try: return server.fetch_lnurl_json(url, timeout=timeout)
+            finally:
+                self.assertEqual(lookup.call_count, 1)
+                self.assertEqual(fixture.connected, [('93.184.216.34', 443)])
+                context.wrap_socket.assert_called_once_with(fixture, server_hostname='provider.example')
+                self.assertIn(b'Host: provider.example\r\n', fixture.sent)
+                self.assertTrue(fixture.closed)
+    def test_redirects_never_connect_to_another_destination(self):
         for destination in ['http://127.0.0.1/admin', 'https://10.0.0.1/private', 'https://other.example/callback']:
             for status in [301, 302, 303, 307, 308]:
-                fixture = FixtureHttps(redirect=destination, status=status)
+                with self.subTest(destination=destination, status=status), self.assertRaisesRegex(ValueError, 'redirects are not permitted'):
-                with self.subTest(destination=destination, status=status):
+                    self.transport(FixtureSocket(redirect=destination, status=status))
-                    with self.assertRaisesRegex(ValueError, 'redirects are not permitted'): self.transport(fixture)
+    def test_dns_rebinding_cannot_trigger_second_lookup_and_tls_keeps_hostname(self):
-                    self.assertEqual(fixture.calls, ['https://provider.example/lnurl'])
+        fixture = FixtureSocket(body=b'{"tag":"payRequest"}')
+        self.assertEqual(self.transport(fixture), {'tag':'payRequest'})
+        self.assertEqual(fixture.timeout, 10)
     def test_response_cap_precedes_json_parsing(self):
-        with self.assertRaisesRegex(ValueError, 'response too large'): self.transport(FixtureHttps(body=b' ' * 65537))
+        with self.assertRaisesRegex(ValueError, 'response too large'): self.transport(FixtureSocket(body=b' ' * 65537))
-    def test_valid_bounded_json(self):
+    def test_private_initial_resolution_never_opens_socket(self):
-        self.assertEqual(self.transport(FixtureHttps(body=b'{"tag":"payRequest"}')), {'tag':'payRequest'})
+        with patch('server.socket.getaddrinfo', return_value=PRIVATE), patch('server.socket.socket') as connect:
-    def test_unsafe_initial_url_never_opens_transport(self):
+            with self.assertRaisesRegex(ValueError, 'public host'): server.fetch_lnurl_json('https://provider.example/lnurl')
-        with patch('server.assert_public_https', side_effect=ValueError('unsafe')), patch('server.build_opener') as opener:
+            connect.assert_not_called()
-            with self.assertRaises(ValueError): server.fetch_lnurl_json('https://127.0.0.1/private')
+    def test_non_https_and_userinfo_never_resolve(self):
-            opener.assert_not_called()
+        for url in ['http://provider.example', 'https://owner:secret@provider.example', 'https://provider.example/#fragment']:
+            with self.subTest(url=url), patch('server.socket.getaddrinfo') as lookup:
+                with self.assertRaises(ValueError): server.fetch_lnurl_json(url)
+                lookup.assert_not_called()
+    def test_screen_metadata_retains_four_second_timeout_and_shared_transport(self):
+        with patch('server.fetch_lnurl_json', return_value={'tag':'payRequest','allowsNostr':True,'nostrPubkey':'a'*64}) as fetch, patch('server.helper', return_value={'tips':[]}):
+            self.assertEqual(server.screen_tips('fixture-npub','merchant@provider.example'), [])
+            fetch.assert_called_once_with('https://provider.example/.well-known/lnurlp/merchant', timeout=4)
+        fixture=FixtureSocket();self.transport(fixture,timeout=4);self.assertEqual(fixture.timeout,4)
     def test_metadata_and_invoice_use_restricted_transport(self):
-        pay = {'tag':'payRequest','callback':'https://provider.example/invoice','minSendable':1,'maxSendable':5000}
+        pay={'tag':'payRequest','callback':'https://provider.example/invoice','minSendable':1,'maxSendable':5000}
-        with patch('server.assert_public_https', side_effect=lambda value:value), patch('server.fetch_lnurl_json', side_effect=[pay,{'pr':'fixture-invoice'}]) as fetch, patch('server.fetch_json') as unrestricted:
+        with patch('server.fetch_lnurl_json', side_effect=[pay,{'pr':'fixture-invoice'}]) as fetch, patch('server.fetch_json') as unrestricted:
             result=server.lightning_invoice('merchant@provider.example',1000)
         self.assertEqual(result['bolt11'],'fixture-invoice');self.assertEqual(fetch.call_count,2);unrestricted.assert_not_called()
     def test_zap_fallback_uses_restricted_transport(self):
         pay={'tag':'payRequest','callback':'https://provider.example/invoice','allowsNostr':True,'nostrPubkey':'a'*64}
-        with patch('server.assert_public_https',side_effect=lambda value:value), patch('server.helper',return_value={'event':{},'pubkey':'b'*64,'lnurl':'fixture-lnurl'}), patch('server.fetch_lnurl_json',side_effect=[pay,HTTPError('',400,'fixture',{},None),{'pr':'fixture-invoice'}]) as fetch, patch('server.fetch_json') as unrestricted:
+        with patch('server.helper',return_value={'event':{},'pubkey':'b'*64,'lnurl':'fixture-lnurl'}), patch('server.fetch_lnurl_json',side_effect=[pay,HTTPError('',400,'fixture',{},None),{'pr':'fixture-invoice'}]) as fetch, patch('server.fetch_json') as unrestricted:
             result=server.lightning_invoice('merchant@provider.example',1000)
         self.assertFalse(result['verifiable']);self.assertEqual(fetch.call_count,3);unrestricted.assert_not_called()