Add pre-migration tests for networking internals

Author: sandhose

tests/test_blacklisting.py

@@ -9,17 +9,62 @@
 
 from unittest.mock import patch
 
+from netaddr import IPAddress, IPSet
 from twisted.internet.error import DNSLookupError
 from twisted.test.proto_helpers import StringTransport
 from twisted.trial.unittest import TestCase
 from twisted.web.client import Agent
 
-from sydent.http.blacklisting_reactor import BlacklistingReactorWrapper
+from sydent.http.blacklisting_reactor import (
+    BlacklistingReactorWrapper,
+    check_against_blacklist,
+)
 from sydent.http.srvresolver import Server
 
 from tests.utils import AsyncMock, make_request, make_sydent
 
 
+class CheckAgainstBlacklistTest(TestCase):
+    """Tests for the check_against_blacklist() utility function."""
+
+    def test_blacklisted_ipv4(self) -> None:
+        blacklist = IPSet(["5.0.0.0/8"])
+        self.assertTrue(check_against_blacklist(IPAddress("5.1.2.3"), None, blacklist))
+
+    def test_not_blacklisted(self) -> None:
+        blacklist = IPSet(["5.0.0.0/8"])
+        self.assertFalse(check_against_blacklist(IPAddress("1.2.3.4"), None, blacklist))
+
+    def test_whitelisted_overrides_blacklist(self) -> None:
+        blacklist = IPSet(["5.0.0.0/8"])
+        whitelist = IPSet(["5.1.1.1"])
+        self.assertFalse(
+            check_against_blacklist(IPAddress("5.1.1.1"), whitelist, blacklist)
+        )
+
+    def test_ipv6_loopback_blocked(self) -> None:
+        blacklist = IPSet(["::1/128"])
+        self.assertTrue(check_against_blacklist(IPAddress("::1"), None, blacklist))
+
+    def test_ipv6_link_local_blocked(self) -> None:
+        blacklist = IPSet(["fe80::/10"])
+        self.assertTrue(check_against_blacklist(IPAddress("fe80::1"), None, blacklist))
+
+    def test_ipv4_mapped_ipv6_blocked(self) -> None:
+        """IPv4-mapped IPv6 addresses (::ffff:127.0.0.1) should be blockable."""
+        blacklist = IPSet(["127.0.0.0/8", "::ffff:127.0.0.0/104"])
+        self.assertTrue(
+            check_against_blacklist(IPAddress("::ffff:127.0.0.1"), None, blacklist)
+        )
+
+    def test_ipv6_whitelist_overrides(self) -> None:
+        blacklist = IPSet(["fe80::/10"])
+        whitelist = IPSet(["fe80::1"])
+        self.assertFalse(
+            check_against_blacklist(IPAddress("fe80::1"), whitelist, blacklist)
+        )
+
+
 class BlacklistingAgentTest(TestCase):
     def setUp(self):
         config = {

tests/test_httpcommon.py

@@ -0,0 +1,181 @@
+# Copyright 2025 New Vector Ltd.
+#
+# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+# Please see LICENSE files in the repository root for full details.
+
+from io import BytesIO
+from unittest.mock import MagicMock
+
+from twisted.internet import defer
+from twisted.python.failure import Failure
+from twisted.trial import unittest
+from twisted.web.client import ResponseDone
+from twisted.web.http import PotentialDataLoss
+from twisted.web.iweb import UNKNOWN_LENGTH
+
+from sydent.http.httpcommon import (
+    BodyExceededMaxSize,
+    SizeLimitingRequest,
+    _DiscardBodyWithMaxSizeProtocol,
+    _ReadBodyWithMaxSizeProtocol,
+    read_body_with_max_size,
+)
+
+
+class ReadBodyWithMaxSizeProtocolTest(unittest.TestCase):
+    """Tests for _ReadBodyWithMaxSizeProtocol."""
+
+    def _make_protocol(
+        self, max_size: int | None = None
+    ) -> tuple[_ReadBodyWithMaxSizeProtocol, "defer.Deferred[bytes]"]:
+        d: defer.Deferred[bytes] = defer.Deferred()
+        protocol = _ReadBodyWithMaxSizeProtocol(d, max_size)
+        protocol.transport = MagicMock()
+        return protocol, d
+
+    def test_reads_body_under_limit(self) -> None:
+        """Body under the limit is read successfully."""
+        protocol, d = self._make_protocol(max_size=100)
+        protocol.dataReceived(b"hello ")
+        protocol.dataReceived(b"world")
+        protocol.connectionLost(Failure(ResponseDone()))
+        self.assertEqual(self.successResultOf(d), b"hello world")
+
+    def test_exceeds_max_size(self) -> None:
+        """Body exceeding max_size triggers BodyExceededMaxSize."""
+        protocol, d = self._make_protocol(max_size=5)
+        protocol.dataReceived(b"too much data")
+        self.failureResultOf(d, BodyExceededMaxSize)
+        protocol.transport.abortConnection.assert_called_once()
+
+    def test_exact_boundary(self) -> None:
+        """Body exactly at max_size triggers the error (>= check)."""
+        protocol, d = self._make_protocol(max_size=5)
+        protocol.dataReceived(b"12345")
+        self.failureResultOf(d, BodyExceededMaxSize)
+
+    def test_no_max_size(self) -> None:
+        """With max_size=None, any amount of data is accepted."""
+        protocol, d = self._make_protocol(max_size=None)
+        protocol.dataReceived(b"x" * 10000)
+        protocol.connectionLost(Failure(ResponseDone()))
+        self.assertEqual(len(self.successResultOf(d)), 10000)
+
+    def test_potential_data_loss_succeeds(self) -> None:
+        """PotentialDataLoss is treated as success (same as ResponseDone)."""
+        protocol, d = self._make_protocol(max_size=1000)
+        protocol.dataReceived(b"partial data")
+        protocol.connectionLost(Failure(PotentialDataLoss()))
+        self.assertEqual(self.successResultOf(d), b"partial data")
+
+    def test_connection_error_propagates(self) -> None:
+        """An unexpected connection loss reason propagates the error."""
+        protocol, d = self._make_protocol(max_size=1000)
+        protocol.dataReceived(b"some data")
+        error = Failure(Exception("connection reset"))
+        protocol.connectionLost(error)
+        f = self.failureResultOf(d)
+        self.assertIsInstance(f.value, Exception)
+
+
+class DiscardBodyWithMaxSizeProtocolTest(unittest.TestCase):
+    """Tests for _DiscardBodyWithMaxSizeProtocol."""
+
+    def test_errors_on_data_received(self) -> None:
+        """Fires errback and aborts connection on first data."""
+        d: defer.Deferred[bytes] = defer.Deferred()
+        protocol = _DiscardBodyWithMaxSizeProtocol(d)
+        protocol.transport = MagicMock()
+        protocol.dataReceived(b"any data")
+        self.failureResultOf(d, BodyExceededMaxSize)
+        protocol.transport.abortConnection.assert_called_once()
+
+    def test_errors_on_connection_lost(self) -> None:
+        """Also fires errback if connectionLost fires first."""
+        d: defer.Deferred[bytes] = defer.Deferred()
+        protocol = _DiscardBodyWithMaxSizeProtocol(d)
+        protocol.transport = MagicMock()
+        protocol.connectionLost(Failure(ResponseDone()))
+        self.failureResultOf(d, BodyExceededMaxSize)
+
+    def test_idempotent(self) -> None:
+        """Multiple calls to _maybe_fail don't double-fire the deferred."""
+        d: defer.Deferred[bytes] = defer.Deferred()
+        protocol = _DiscardBodyWithMaxSizeProtocol(d)
+        protocol.transport = MagicMock()
+        protocol.dataReceived(b"first")
+        protocol.dataReceived(b"second")  # should not raise
+        protocol.connectionLost(Failure(ResponseDone()))
+        self.failureResultOf(d, BodyExceededMaxSize)
+
+
+class ReadBodyWithMaxSizeFunctionTest(unittest.TestCase):
+    """Tests for the read_body_with_max_size() top-level function."""
+
+    def _make_response(self, length: int | object = UNKNOWN_LENGTH) -> MagicMock:
+        response = MagicMock()
+        response.length = length
+        return response
+
+    def test_content_length_exceeds_limit_uses_discard(self) -> None:
+        """When Content-Length > max_size, the Discard protocol is used."""
+        response = self._make_response(length=200)
+        read_body_with_max_size(response, max_size=100)
+        response.deliverBody.assert_called_once()
+        protocol = response.deliverBody.call_args[0][0]
+        self.assertIsInstance(protocol, _DiscardBodyWithMaxSizeProtocol)
+
+    def test_content_length_under_limit_uses_reader(self) -> None:
+        """When Content-Length <= max_size, the Read protocol is used."""
+        response = self._make_response(length=50)
+        read_body_with_max_size(response, max_size=100)
+        response.deliverBody.assert_called_once()
+        protocol = response.deliverBody.call_args[0][0]
+        self.assertIsInstance(protocol, _ReadBodyWithMaxSizeProtocol)
+
+    def test_unknown_length_uses_reader(self) -> None:
+        """When Content-Length is unknown, the Read protocol is used."""
+        response = self._make_response(length=UNKNOWN_LENGTH)
+        read_body_with_max_size(response, max_size=100)
+        response.deliverBody.assert_called_once()
+        protocol = response.deliverBody.call_args[0][0]
+        self.assertIsInstance(protocol, _ReadBodyWithMaxSizeProtocol)
+
+    def test_no_max_size_uses_reader(self) -> None:
+        """When max_size is None, always uses the Read protocol."""
+        response = self._make_response(length=999999)
+        read_body_with_max_size(response, max_size=None)
+        response.deliverBody.assert_called_once()
+        protocol = response.deliverBody.call_args[0][0]
+        self.assertIsInstance(protocol, _ReadBodyWithMaxSizeProtocol)
+
+
+class SizeLimitingRequestTest(unittest.TestCase):
+    """Tests for SizeLimitingRequest.handleContentChunk()."""
+
+    def _make_request(self) -> SizeLimitingRequest:
+        """Create a minimal SizeLimitingRequest for testing."""
+        req = SizeLimitingRequest.__new__(SizeLimitingRequest)
+        req.content = BytesIO()
+        req.transport = MagicMock()
+        # The client attribute is accessed for logging.
+        req.client = MagicMock()
+        return req
+
+    def test_accepts_data_under_limit(self) -> None:
+        """Chunks totalling under MAX_REQUEST_SIZE are accepted."""
+        req = self._make_request()
+        data = b"x" * 1000
+        req.handleContentChunk(data)
+        self.assertEqual(req.content.tell(), 1000)
+        req.transport.abortConnection.assert_not_called()
+
+    def test_aborts_on_oversize(self) -> None:
+        """Connection is aborted when cumulative data exceeds MAX_REQUEST_SIZE."""
+        req = self._make_request()
+        # Write data right up to the limit.
+        req.content.write(b"x" * (512 * 1024))
+        req.content.seek(512 * 1024)
+        # The next chunk pushes over.
+        req.handleContentChunk(b"x")
+        req.transport.abortConnection.assert_called_once()

tests/test_replication.py

@@ -1,6 +1,7 @@
 import json
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
+from OpenSSL import crypto
 from twisted.internet import defer
 from twisted.trial import unittest
 from twisted.web.client import Response
@@ -190,3 +191,101 @@ def request(method, uri, headers, body):
             # will push will be 1, so we need to subtract 1 when figuring out which index
             # to lookup.
             self.assertDictEqual(assoc, signed_assocs[int(assoc_id) - 1])
+
+
+class ReplicationCNTest(unittest.TestCase):
+    """Tests for peer certificate CN extraction edge cases in the replication servlet."""
+
+    def setUp(self) -> None:
+        self.sydent = make_sydent()
+
+        # Insert a known peer.
+        cur = self.sydent.db.cursor()
+        cur.execute(
+            "INSERT INTO peers (name, port, lastSentVersion, active) VALUES (?, ?, ?, ?)",
+            ("fake.server", 1234, 0, 1),
+        )
+        peer_public_key_base64 = "+vB8mTaooD/MA8YYZM8t9+vnGhP1937q2icrqPV9JTs"
+        cur.execute(
+            "INSERT INTO peer_pubkeys (peername, alg, key) VALUES (?, ?, ?)",
+            ("fake.server", "ed25519", peer_public_key_base64),
+        )
+        self.sydent.db.commit()
+
+    def test_known_peer_cn_accepted(self) -> None:
+        """A peer cert with CN matching a known peer is accepted (existing test validates this,
+        but let's have a focused unit-level check)."""
+        self.sydent.run()
+
+        # The FakeChannel.getPeerCertificate() returns a cert with CN=fake.server,
+        # and we inserted fake.server as a peer. A valid request should be accepted.
+        body = {"sgAssocs": {}}
+        request, channel = make_request(
+            self.sydent.reactor,
+            self.sydent.replicationHttpsServer.factory,
+            "POST",
+            "/_matrix/identity/replicate/v1/push",
+            body,
+        )
+        self.assertEqual(channel.code, 200)
+
+    def test_unknown_peer_cn_rejected(self) -> None:
+        """A peer cert with CN that doesn't match any known peer returns 403."""
+        self.sydent.run()
+
+        # Generate a cert with a CN that is NOT in the peers table.
+        unknown_key = crypto.PKey()
+        unknown_key.generate_key(crypto.TYPE_RSA, 2048)
+        unknown_cert = crypto.X509()
+        unknown_cert.get_subject().CN = "unknown.server"
+        unknown_cert.set_serial_number(1000)
+        unknown_cert.gmtime_adj_notBefore(0)
+        unknown_cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)
+        unknown_cert.set_issuer(unknown_cert.get_subject())
+        unknown_cert.set_pubkey(unknown_key)
+        unknown_cert.sign(unknown_key, "sha256")
+
+        # Patch FakeChannel.getPeerCertificate to return our unknown cert.
+        with patch(
+            "tests.utils.FakeChannel.getPeerCertificate", return_value=unknown_cert
+        ):
+            body = {"sgAssocs": {}}
+            request, channel = make_request(
+                self.sydent.reactor,
+                self.sydent.replicationHttpsServer.factory,
+                "POST",
+                "/_matrix/identity/replicate/v1/push",
+                body,
+            )
+        self.assertEqual(channel.code, 403)
+        self.assertEqual(channel.json_body["errcode"], "M_UNKNOWN_PEER")
+
+    def test_no_cn_rejected(self) -> None:
+        """A peer cert with no commonName returns 403."""
+        self.sydent.run()
+
+        # Generate a cert with no CN set.
+        no_cn_key = crypto.PKey()
+        no_cn_key.generate_key(crypto.TYPE_RSA, 2048)
+        no_cn_cert = crypto.X509()
+        # Don't set CN — leave subject empty.
+        no_cn_cert.set_serial_number(2000)
+        no_cn_cert.gmtime_adj_notBefore(0)
+        no_cn_cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)
+        no_cn_cert.set_issuer(no_cn_cert.get_subject())
+        no_cn_cert.set_pubkey(no_cn_key)
+        no_cn_cert.sign(no_cn_key, "sha256")
+
+        with patch(
+            "tests.utils.FakeChannel.getPeerCertificate", return_value=no_cn_cert
+        ):
+            body = {"sgAssocs": {}}
+            request, channel = make_request(
+                self.sydent.reactor,
+                self.sydent.replicationHttpsServer.factory,
+                "POST",
+                "/_matrix/identity/replicate/v1/push",
+                body,
+            )
+        self.assertEqual(channel.code, 403)
+        self.assertEqual(channel.json_body["errcode"], "M_UNKNOWN_PEER")