1 files changed, 87 insertions, 7 deletions

diff --git a/packages/meshbay-node/tests/test_linkpreview.py b/packages/meshbay-node/tests/test_linkpreview.py
index 9fca186..3e6eaf7 100644
--- a/packages/meshbay-node/tests/test_linkpreview.py
+++ b/packages/meshbay-node/tests/test_linkpreview.py
@@ -6,12 +6,13 @@ decides an outbound request from the operator's machine. Anything that is not
 a public http(s) address must be refused before a socket opens.
 """
 
+import asyncio
 import socket
 
 import httpx
 import pytest
 from meshbay_node import linkpreview
-from meshbay_node.linkpreview import UnsafeURL, safe_url
+from meshbay_node.linkpreview import UnsafeURL, check_url, safe_url
 
 PUBLIC_IP = "93.184.216.34"       # example.com, historically
 
@@ -43,9 +44,9 @@ def resolves_public(monkeypatch):
     "javascript:alert(1)",
     "not a url",
 ])
-def test_safe_url_refuses(url):
+async def test_check_url_refuses(url):
     with pytest.raises(UnsafeURL):
-        safe_url(url)
+        await check_url(url)
 
 
 @pytest.mark.parametrize("url", [
@@ -72,11 +73,11 @@ def test_safe_url_allows_the_web_ports(url, resolves_public):
     assert safe_url(url) == url
 
 
-def test_safe_url_accepts_a_public_host(resolves_public):
-    assert safe_url("https://example.com/some/page") == "https://example.com/some/page"
+async def test_check_url_accepts_a_public_host(resolves_public):
+    assert await check_url("https://example.com/some/page") == "https://example.com/some/page"
 
 
-def test_safe_url_refuses_a_host_with_any_private_record(monkeypatch):
+async def test_check_url_refuses_a_host_with_any_private_record(monkeypatch):
     def mixed(host, port, *a, **k):
         return [
             (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (PUBLIC_IP, port)),
@@ -84,7 +85,7 @@ def test_safe_url_refuses_a_host_with_any_private_record(monkeypatch):
         ]
     monkeypatch.setattr(linkpreview.socket, "getaddrinfo", mixed)
     with pytest.raises(UnsafeURL):
-        safe_url("https://sneaky.example/x")
+        await check_url("https://sneaky.example/x")
 
 
 # ── fetch_preview ──────────────────────────────────────────────────────────
@@ -273,3 +274,82 @@ async def test_a_declared_oversized_image_is_not_read(resolves_public):
     async with _client(handler) as c:
         assert await linkpreview.fetch_image("https://example.com/x.png", client=c) is None
     assert counter["sent"] <= _CHUNK
+
+
+# ── The connection goes where the check said ───────────────────────────────
+
+class _Recorder:
+    """Stands in for the real socket layer under the pinned backend."""
+    def __init__(self):
+        self.hosts = []
+
+    async def connect_tcp(self, host, port, **kw):
+        self.hosts.append(host)
+        raise httpx.ConnectError("recorded, not connected")
+
+
+def _pinned_with(recorder):
+    backend = linkpreview._PinnedBackend()
+    backend._inner = recorder
+    return backend
+
+
+async def test_the_socket_is_opened_to_the_checked_address(resolves_public):
+    rec = _Recorder()
+    with pytest.raises(httpx.ConnectError):
+        await _pinned_with(rec).connect_tcp("example.com", 443)
+    assert rec.hosts == [PUBLIC_IP], "the name, not the checked address, was dialled"
+
+
+async def test_a_name_that_rebinds_never_reaches_the_lan(monkeypatch):
+    """
+    Answers clean when checked, then with a LAN address. Checked once and
+    dialled by name, the request would go to the LAN before anything looked;
+    resolved and checked by the backend that dials, it goes nowhere.
+    """
+    answers = iter([PUBLIC_IP, "192.168.1.1", "192.168.1.1"])
+
+    def rebinding(host, port, *a, **k):
+        return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "",
+                 (next(answers), port))]
+    monkeypatch.setattr(linkpreview.socket, "getaddrinfo", rebinding)
+
+    await check_url("http://rebind.example/x")          # the clean answer
+    rec = _Recorder()
+    with pytest.raises(UnsafeURL):
+        await _pinned_with(rec).connect_tcp("rebind.example", 80)
+    assert rec.hosts == []
+
+
+async def test_the_real_client_is_pinned():
+    """What `fetch_preview` uses when the caller gives no client."""
+    client = linkpreview._new_client()
+    try:
+        pool = client._transport._pool
+        assert isinstance(pool._network_backend, linkpreview._PinnedBackend)
+        assert client._trust_env is False, "a proxy from the environment would unpin it"
+    finally:
+        await client.aclose()
+
+
+async def test_resolving_does_not_hold_the_event_loop(monkeypatch):
+    import time as _time
+
+    def slow(host, port, *a, **k):
+        _time.sleep(0.4)
+        return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (PUBLIC_IP, port))]
+    monkeypatch.setattr(linkpreview.socket, "getaddrinfo", slow)
+
+    ticks = 0
+
+    async def ticker():
+        nonlocal ticks
+        while True:
+            await asyncio.sleep(0.02)
+            ticks += 1
+    t = asyncio.create_task(ticker())
+    try:
+        await check_url("https://slow.example/x")
+    finally:
+        t.cancel()
+    assert ticks >= 10, "the event loop stood still while a name resolved"