"""
`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)
@pytest.mark.parametrize("url", [
"http://example.com:22/x", # SSH
"http://example.com:3306/x", # MySQL
"http://example.com:6379/x", # Redis
"http://example.com:9200/x", # Elasticsearch
"http://example.com:5000/x", # a common internal admin port
])
def test_safe_url_refuses_non_web_ports(url, resolves_public):
with pytest.raises(UnsafeURL):
safe_url(url)
@pytest.mark.parametrize("url", [
"http://example.com/x", # implicit 80
"https://example.com/x", # implicit 443
"http://example.com:80/x",
"https://example.com:443/x",
"http://example.com:8080/x",
"https://example.com:8443/x",
])
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"
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 = """
Fallback Title
...body we should not need...
"""
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="Just A Title")
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="hi")
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
async def test_fetch_image_refuses_a_decompression_bomb(resolves_public, monkeypatch):
from io import BytesIO
from PIL import Image
# A tiny file that reports enormous dimensions from its header alone.
monkeypatch.setattr(linkpreview, "_MAX_IMAGE_PIXELS", 1_000_000)
buf = BytesIO()
Image.new("RGB", (2000, 2000), (0, 0, 0)).save(buf, format="PNG") # 4 MP > cap
bomb = buf.getvalue()
def handler(request):
return httpx.Response(200, headers={"content-type": "image/png"}, content=bomb)
async with _client(handler) as c:
assert await linkpreview.fetch_image("https://example.com/x.png", client=c) is None