path: root/packages/meshbay-node/tests/test_linkpreview.py

"""
`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 asyncio
import socket

import httpx
import pytest
from meshbay_node import linkpreview
from meshbay_node.linkpreview import UnsafeURL, check_url, 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",
])
async def test_check_url_refuses(url):
    with pytest.raises(UnsafeURL):
        await check_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


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"


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)),
            (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", port)),
        ]
    monkeypatch.setattr(linkpreview.socket, "getaddrinfo", mixed)
    with pytest.raises(UnsafeURL):
        await check_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


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


# ── Size caps bind while reading, not after ─────────────────────────────────
#
# `client.get` read and decoded the whole body before the caps looked at it, so
# a link posted in chat could make the node hold any amount of data. These
# count what the server actually had to hand over.

_CHUNK = 64 * 1024


def _endless(counter, head=b""):
    async def body():
        if head:
            counter["sent"] += len(head)
            yield head
        for _ in range(2000):                    # 125 MiB if nobody stops
            counter["sent"] += _CHUNK
            yield b"x" * _CHUNK
    return body()


async def test_a_huge_page_is_not_read_past_the_cap(resolves_public):
    counter = {"sent": 0}
    head = b"<html><head><title>T</title></head><body>"

    def handler(request):
        return httpx.Response(200, headers={"content-type": "text/html"},
                              content=_endless(counter, head))
    async with _client(handler) as c:
        meta = await linkpreview.fetch_preview("https://example.com/", client=c)
    assert meta is not None and meta["title"] == "T"
    assert counter["sent"] <= linkpreview._MAX_HTML_BYTES + 2 * _CHUNK


async def test_a_huge_image_is_refused_without_being_read(resolves_public):
    counter = {"sent": 0}

    def handler(request):
        return httpx.Response(200, headers={"content-type": "image/png"},
                              content=_endless(counter))
    async with _client(handler) as c:
        assert await linkpreview.fetch_image("https://example.com/x.png", client=c) is None
    assert counter["sent"] <= linkpreview._MAX_IMAGE_BYTES + 2 * _CHUNK


async def test_a_compressed_body_is_capped_after_decoding(resolves_public):
    import zlib
    comp = zlib.compressobj(9, zlib.DEFLATED, 31)          # gzip
    counter = {"inflated": 0}

    async def body():
        yield comp.compress(b"<html><head><title>T</title></head><body>")
        for _ in range(2000):                               # 125 MiB inflated
            counter["inflated"] += _CHUNK
            # Flushed per chunk, so what is counted is what went on the wire.
            yield comp.compress(b"x" * _CHUNK) + comp.flush(zlib.Z_SYNC_FLUSH)
        yield comp.flush()

    def handler(request):
        return httpx.Response(200, headers={"content-type": "text/html",
                                            "content-encoding": "gzip"},
                              content=body())
    async with _client(handler) as c:
        meta = await linkpreview.fetch_preview("https://example.com/", client=c)
    assert meta is not None
    assert counter["inflated"] < 50 * _CHUNK      # stopped early, not at 125 MiB


async def test_a_declared_oversized_image_is_not_read(resolves_public):
    counter = {"sent": 0}

    def handler(request):
        return httpx.Response(
            200, headers={"content-type": "image/png",
                          "content-length": str(linkpreview._MAX_IMAGE_BYTES + 1)},
            content=_endless(counter))
    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"