Merge branch 'feat/chat-link-previews'

author: Christophe Besson <cbesson@gmail.com> 2026-08-28 03:43:29 +0200
committer: Christophe Besson <cbesson@gmail.com> 2026-08-28 03:43:29 +0200
commit: 25ebaed11772056c611c3ce650387d6961e0baa0 (patch)
tree: e22bee25a2cdf3e11cbfce2525728d2010ab85d8 /packages/meshbay-node/tests/test_linkpreview.py
parent: 2f573b4143212b21d92ff20179356acb8f8fe635 (diff)
parent: ce4e10c4b8bd9c66c375c3a5d5c18d8552655775 (diff)
download: meshbay-25ebaed11772056c611c3ce650387d6961e0baa0.tar.gz
1 files changed, 155 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_linkpreview.py b/packages/meshbay-node/tests/test_linkpreview.py
new file mode 100644
index 0000000..0dd951a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_linkpreview.py
@@ -0,0 +1,155 @@
+"""
+`linkpreview` — the SSRF gate and the OpenGraph parse.
+
+The gate is the part with teeth: the URL is chosen by a *member*, and it
+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 socket
+
+import httpx
+import pytest
+from meshbay_node import linkpreview
+from meshbay_node.linkpreview import UnsafeURL, safe_url
+
+PUBLIC_IP = "93.184.216.34"       # example.com, historically
+
+
+@pytest.fixture
+def resolves_public(monkeypatch):
+    """Every hostname resolves to one public address."""
+    def fake_getaddrinfo(host, port, *a, **k):
+        return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "",
+                 (PUBLIC_IP, port or 80))]
+    monkeypatch.setattr(linkpreview.socket, "getaddrinfo", fake_getaddrinfo)
+
+
+# ── safe_url ────────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("url", [
+    "http://127.0.0.1/x",
+    "http://localhost/x",            # resolves to loopback on any box
+    "http://169.254.169.254/latest/meta-data/",   # cloud metadata
+    "http://[::1]/x",
+    "http://10.1.2.3/x",
+    "http://192.168.0.1/x",
+    "http://172.16.0.1/x",
+    "http://0.0.0.0/x",
+    "http://[::ffff:127.0.0.1]/x",   # v4-mapped loopback
+    "ftp://example.com/x",
+    "file:///etc/passwd",
+    "http://user:pass@example.com/x",
+    "javascript:alert(1)",
+    "not a url",
+])
+def test_safe_url_refuses(url):
+    with pytest.raises(UnsafeURL):
+        safe_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"
+
+
+def test_safe_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)),
+            (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", port)),
+        ]
+    monkeypatch.setattr(linkpreview.socket, "getaddrinfo", mixed)
+    with pytest.raises(UnsafeURL):
+        safe_url("https://sneaky.example/x")
+
+
+# ── fetch_preview ──────────────────────────────────────────────────────────
+
+_HTML = """
+<!doctype html><html><head>
+  <title>Fallback Title</title>
+  <meta property="og:title" content="The Real Title">
+  <meta property="og:description" content="A short summary of the page.">
+  <meta property="og:site_name" content="Example">
+  <meta property="og:image" content="/card.png">
+  <meta name="description" content="ignored, og wins">
+</head><body>...body we should not need...</body></html>
+"""
+
+
+def _client(handler):
+    return httpx.AsyncClient(transport=httpx.MockTransport(handler),
+                             timeout=5.0, max_redirects=0)
+
+
+async def test_fetch_preview_reads_opengraph(resolves_public):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "text/html; charset=utf-8"},
+                              text=_HTML)
+    async with _client(handler) as c:
+        meta = await linkpreview.fetch_preview("https://example.com/article", client=c)
+    assert meta["title"] == "The Real Title"
+    assert meta["description"] == "A short summary of the page."
+    assert meta["site_name"] == "Example"
+    assert meta["image_url"] == "https://example.com/card.png"   # absolutised
+
+
+async def test_fetch_preview_falls_back_to_title_tag(resolves_public):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "text/html"},
+                              text="<html><head><title>Just A Title</title></head></html>")
+    async with _client(handler) as c:
+        meta = await linkpreview.fetch_preview("https://example.com/", client=c)
+    assert meta["title"] == "Just A Title"
+    assert meta["description"] is None
+
+
+async def test_fetch_preview_gives_up_on_non_html(resolves_public):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "application/pdf"},
+                              content=b"%PDF-1.4")
+    async with _client(handler) as c:
+        assert await linkpreview.fetch_preview("https://example.com/x.pdf", client=c) is None
+
+
+async def test_fetch_preview_gives_up_when_nothing_worth_showing(resolves_public):
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "text/html"},
+                              text="<html><head></head><body>hi</body></html>")
+    async with _client(handler) as c:
+        assert await linkpreview.fetch_preview("https://example.com/", client=c) is None
+
+
+async def test_fetch_preview_revalidates_redirects(monkeypatch):
+    # First host is public; it 302s to a loopback address.
+    calls = {"n": 0}
+
+    def resolve(host, port, *a, **k):
+        ip = PUBLIC_IP if host == "ok.example" else "127.0.0.1"
+        return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (ip, port or 80))]
+    monkeypatch.setattr(linkpreview.socket, "getaddrinfo", resolve)
+
+    def handler(request):
+        calls["n"] += 1
+        return httpx.Response(302, headers={"location": "http://internal.example/secret"})
+    async with _client(handler) as c:
+        meta = await linkpreview.fetch_preview("https://ok.example/start", client=c)
+    assert meta is None
+    assert calls["n"] == 1          # stopped at the redirect, never fetched internal
+
+
+async def test_fetch_image_downscales(resolves_public):
+    from io import BytesIO
+
+    from PIL import Image
+    buf = BytesIO()
+    Image.new("RGB", (2000, 1500), (10, 20, 30)).save(buf, format="PNG")
+    big_png = buf.getvalue()
+
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "image/png"}, content=big_png)
+    async with _client(handler) as c:
+        jpeg = await linkpreview.fetch_image("https://example.com/card.png", client=c)
+    assert jpeg and jpeg[:2] == b"\xff\xd8"        # JPEG SOI
+    with Image.open(BytesIO(jpeg)) as im:
+        assert max(im.size) <= linkpreview._IMAGE_MAX_DIM
author	Christophe Besson <cbesson@gmail.com>	2026-08-28 03:43:29 +0200
committer	Christophe Besson <cbesson@gmail.com>	2026-08-28 03:43:29 +0200
commit	25ebaed11772056c611c3ce650387d6961e0baa0 (patch)
tree	e22bee25a2cdf3e11cbfce2525728d2010ab85d8 /packages/meshbay-node/tests/test_linkpreview.py
parent	2f573b4143212b21d92ff20179356acb8f8fe635 (diff)
parent	ce4e10c4b8bd9c66c375c3a5d5c18d8552655775 (diff)
download	meshbay-25ebaed11772056c611c3ce650387d6961e0baa0.tar.gz