2 files changed, 154 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_linkpreview.py b/packages/meshbay-node/tests/test_linkpreview.py
index a14b173..9fca186 100644
--- a/packages/meshbay-node/tests/test_linkpreview.py
+++ b/packages/meshbay-node/tests/test_linkpreview.py
@@ -193,3 +193,83 @@ async def test_fetch_image_refuses_a_decompression_bomb(resolves_public, monkeyp
         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
diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py
index 5c94855..585a909 100644
--- a/packages/meshbay-node/tests/test_roster_pairing.py
+++ b/packages/meshbay-node/tests/test_roster_pairing.py
@@ -339,6 +339,80 @@ async def test_failures_are_counted_across_connections(tmp_path, roster):
         "reconnecting must not reset the pairing budget")
 
 
+async def _grind_wrong_codes(tmp_path, roster, ctx, accounts, per_account):
+    """Wrong codes from several accounts, each on connections of its own."""
+    for n in range(accounts):
+        uid = f"guesser{n}"
+        for _ in range(per_account):
+            session = _session(tmp_path, roster, user_id=uid)
+            session._ctx = ctx
+            sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+            await session._do_join_request(
+                _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64,
+                          code="AAAA-AAAA", user_id=uid))
+
+
+async def test_a_full_code_lock_never_refuses_a_known_member(tmp_path, roster):
+    """
+    Every member reconnecting gets the group key through join_request, so a
+    lock on wrong codes that also refused *recognised* devices let one member
+    take the node away from everybody.
+    """
+    GROUP = "b" * 32
+    gek = generate_gek()
+    sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full()
+    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
+    await roster.set_member(GROUP, "grenet", ROLE_MEMBER, "active", "cbesson")
+
+    member = _session(tmp_path, roster, user_id="grenet", group_id=GROUP,
+                      gek=gek, join_policy="invite")
+    await _grind_wrong_codes(tmp_path, roster, member._ctx, accounts=6,
+                             per_account=5)
+    assert len(member._ctx["join_failures"]) >= 20
+
+    await member._do_join_request(
+        _join_msg(member, sk_ed, pk_ed_b64, pk_x_b64, user_id="grenet"))
+    reply = _last(member)
+    assert reply.get("ok") is True and reply.get("gek") is True, reply
+    assert unwrap_gek_aes(reply, *_x_raw(sk_x, pk_x_b64)) == gek
+
+
+async def test_ordinary_refusals_do_not_feed_the_code_lock(tmp_path, roster):
+    """`code_required` is what every first contact without a code hears."""
+    ctx = None
+    for n in range(30):
+        session = _session(tmp_path, roster, user_id=f"newcomer{n}")
+        if ctx is None:
+            ctx = session._ctx
+        session._ctx = ctx
+        sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+        await session._do_join_request(
+            _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id=f"newcomer{n}"))
+        assert _last(session).get("reason") == "code_required"
+    assert not ctx.get("join_failures")
+
+
+async def test_one_account_guessing_locks_itself_not_the_others(tmp_path, roster):
+    session = _session(tmp_path, roster)
+    ctx = session._ctx
+    await _grind_wrong_codes(tmp_path, roster, ctx, accounts=1, per_account=6)
+    guesser = _session(tmp_path, roster, user_id="guesser0")
+    guesser._ctx = ctx
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await guesser._do_join_request(
+        _join_msg(guesser, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA",
+                  user_id="guesser0"))
+    assert _last(guesser).get("detail") == "Pairing temporarily locked"
+
+    code = await roster.create_invite("", "grenet", ROLE_OPERATOR, "local-cli")
+    honest = _session(tmp_path, roster, user_id="grenet")
+    honest._ctx = ctx
+    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
+    await honest._do_join_request(
+        _join_msg(honest, sk_ed, pk_ed_b64, pk_x_b64, code=code))
+    assert await roster.find_device("grenet", pk_ed_b64) is not None
+
+
 async def test_group_id_cannot_name_another_group(tmp_path, roster):
     session = _session(tmp_path, roster)
     session._group_id = "a" * 32