Support matrix_room_id in PFS info response

This changes the is-pfs-on-the-same-federation check to use the new
`matrix_room_id` field from the PFS info response.

See raiden-network/raiden-services#858 for details

Author: ulope

raiden/network/pathfinding.py

@@ -75,6 +75,7 @@ class PFSInfo:
     version: str
     confirmed_block_number: BlockNumber
     matrix_server: str
+    matrix_room_id: Optional[str]
 
 
 @dataclass
@@ -164,6 +165,7 @@ def get_pfs_info(url: str) -> PFSInfo:
             version=infos["version"],
             confirmed_block_number=infos["network_info"]["confirmed_block"]["number"],
             matrix_server=matrix_server_info.netloc,
+            matrix_room_id=infos.get("matrix_room_id"),
         )
     except RequestException as e:
         msg = "Selected Pathfinding Service did not respond"
@@ -248,7 +250,6 @@ def configure_pfs_or_exit(
     node_network_id: ChainID,
     token_network_registry_address: TokenNetworkRegistryAddress,
     pathfinding_max_fee: TokenAmount,
-    matrix_servers: List[str],
 ) -> PFSInfo:
     """
     Take in the given pfs_address argument, the service registry and find out a
@@ -315,17 +316,6 @@ def configure_pfs_or_exit(
             f"Raiden will shut down. Please choose a different Pathfinding Service."
         )
 
-    server_in_federation = any(
-        pathfinding_service_info.matrix_server in matrix_server for matrix_server in matrix_servers
-    )
-    # Only check if PFS is right federation when matrix server is not given explicitely
-    if len(matrix_servers) > 0 and not server_in_federation:
-        raise RaidenError(
-            f"The Pathfinding Service {pfs_url} is not connected to the same matrix federation. "
-            f"Please check your settings for PFS and matrix server, if manually chosen. "
-            f"Otherwise, check your environment-type."
-        )
-
     click.secho(
         f"You have chosen the Pathfinding Service at {pfs_url}.\n"
         f"Operator: {pathfinding_service_info.operator}, "
@@ -341,6 +331,48 @@ def configure_pfs_or_exit(
     return pathfinding_service_info
 
 
+def check_pfs_transport_configuration(
+    pfs_info: PFSInfo,
+    pfs_was_autoselected: bool,
+    transport_pfs_broadcast_room_id: str,
+    matrix_server_url: str,
+    matrix_server_was_autoselected: bool,
+) -> None:
+    if pfs_info.matrix_room_id is None:
+        # Special case until all PFSs are upgraded to >= 0.12.0
+        log.warning(
+            "Can't check PFS transport configuration",
+            pfs_version=pfs_info.version,
+            min_required_version="0.12.0",
+            pfs_url=pfs_info.url,
+        )
+        return
+    if pfs_info.matrix_room_id != transport_pfs_broadcast_room_id:
+        msg = (
+            f"The Pathfinding Service at {pfs_info.url} is not connected to the "
+            f"same Matrix transport federation as this Raiden node."
+        )
+        if matrix_server_was_autoselected and pfs_was_autoselected:
+            msg += (
+                f"\nBoth the Matrix transport server at {matrix_server_url} and the PFS were "
+                f"automatically selected. This points to a server-side misconfiguration. "
+                f"Please report this issue at "
+                f"https://github.com/raiden-network/raiden/issues/new?template=bug_report.md"
+            )
+        else:
+            msg += (
+                "\nPlease verify that the Matrix transport server and PFS service you selected "
+                "are intended to be used together or use the automatic server selection."
+            )
+        msg += (
+            f"\n\n"
+            f"- Transport server PFS broadcast room-id: {transport_pfs_broadcast_room_id}\n"
+            f"- PFS server broadcast room-id: {pfs_info.matrix_room_id}"
+        )
+
+        raise RaidenError(msg)
+
+
 def check_pfs_for_production(
     service_registry: Optional[ServiceRegistry], pfs_info: PFSInfo
 ) -> None:

raiden/network/transport/matrix/transport.py

@@ -384,15 +384,15 @@ def _http_retry_delay() -> Iterable[float]:
             None, self._handle_web_rtc_messages, self._handle_sdp_callback
         )
 
-        self._server_url = self._client.api.base_url
-        self._server_name = urlparse(self._server_url).netloc
+        self.server_url = self._client.api.base_url
+        self._server_name = urlparse(self.server_url).netloc
 
         self.greenlets: List[gevent.Greenlet] = list()
 
         self._address_to_retrier: Dict[Address, _RetryQueue] = dict()
         self._displayname_cache = DisplayNameCache()
 
-        self._broadcast_rooms: Dict[str, Room] = dict()
+        self.broadcast_rooms: Dict[str, Room] = dict()
         self._broadcast_queue: JoinableQueue[Tuple[str, Message]] = JoinableQueue()
 
         self._started = False
@@ -734,11 +734,11 @@ def _broadcast(room_name: str, serialized_message: str) -> None:
                     f"Known public rooms: {self._config.broadcast_rooms}."
                 )
             room_name = make_room_alias(self.chain_id, room_name)
-            if room_name not in self._broadcast_rooms:
+            if room_name not in self.broadcast_rooms:
                 room = join_broadcast_room(self._client, f"#{room_name}:{self._server_name}")
-                self._broadcast_rooms[room_name] = room
+                self.broadcast_rooms[room_name] = room
 
-            existing_room = self._broadcast_rooms.get(room_name)
+            existing_room = self.broadcast_rooms.get(room_name)
             assert existing_room, f"Unknown broadcast room: {room_name!r}"
 
             self.log.debug(
@@ -803,7 +803,7 @@ def _initialize_first_sync(self) -> None:
         # Call sync to fetch the inventory rooms and new invites, the sync
         # limit prevents fetching the messages.
         filter_id = self._client.create_sync_filter(
-            not_rooms=self._broadcast_rooms.values(), limit=0
+            not_rooms=self.broadcast_rooms.values(), limit=0
         )
         prev_sync_filter_id = self._client.set_sync_filter_id(filter_id)
         # Need to reset this here, otherwise we might run into problems after a restart
@@ -883,14 +883,14 @@ def _join_broadcast_room(transport: MatrixTransport, room_name: str) -> None:
             transport.log.debug(
                 "Joining broadcast room", broadcast_room_alias=broadcast_room_alias
             )
-            transport._broadcast_rooms[room_name] = join_broadcast_room(
+            transport.broadcast_rooms[room_name] = join_broadcast_room(
                 client=transport._client, broadcast_room_alias=broadcast_room_alias
             )
 
         for suffix in self._config.broadcast_rooms:
             alias_prefix = make_room_alias(self.chain_id, suffix)
 
-            if alias_prefix not in self._broadcast_rooms:
+            if alias_prefix not in self.broadcast_rooms:
                 pool.apply_async(_join_broadcast_room, args=(self, alias_prefix))
 
         pool.join(raise_error=True)
@@ -908,10 +908,10 @@ def _initialize_sync(self) -> None:
             "sync thread, since that is necessary to properly generate the "
             "filters."
         )
-        assert self._broadcast_rooms, msg
+        assert self.broadcast_rooms, msg
 
         broadcast_filter_id = self._client.create_sync_filter(
-            not_rooms=self._broadcast_rooms.values()
+            not_rooms=self.broadcast_rooms.values()
         )
         self._client.set_sync_filter_id(broadcast_filter_id)
 
@@ -1085,7 +1085,7 @@ def _handle_member_join(self, room: Room) -> None:
         if self._is_broadcast_room(room):
             raise AssertionError(
                 f"Broadcast room events should be filtered in syncs: {room.canonical_alias}."
-                f"Joined Broadcast Rooms: {list(self._broadcast_rooms.keys())}"
+                f"Joined Broadcast Rooms: {list(self.broadcast_rooms.keys())}"
                 f"Should be joined to: {self._config.broadcast_rooms}"
             )
 

raiden/tests/integration/network/test_pathfinding.py

@@ -1,3 +1,5 @@
+import dataclasses
+import logging
 from unittest.mock import patch
 
 import pytest
@@ -14,6 +16,7 @@
 from raiden.network.pathfinding import (
     PFSInfo,
     check_pfs_for_production,
+    check_pfs_transport_configuration,
     configure_pfs_or_exit,
     session,
 )
@@ -51,6 +54,7 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
         "version": "0.0.1",
         "payment_address": to_checksum_address(privatekey_to_address(private_keys[0])),
         "matrix_server": "http://matrix.example.com",
+        "matrix_room_id": "!room-id:matrix.example.com",
     }
 
     response = mocked_json_response(response_data=json_data)
@@ -64,7 +68,6 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
             node_network_id=chain_id,
             token_network_registry_address=token_network_registry_address_test_default,
             pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-            matrix_servers=["http://matrix.example.com"],
         )
 
     # With private routing configure_pfs should raise assertion
@@ -76,7 +79,6 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
             node_network_id=chain_id,
             token_network_registry_address=token_network_registry_address_test_default,
             pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-            matrix_servers=["http://matrix.example.com"],
         )
 
     # Asking for auto address
@@ -90,7 +92,6 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
             node_network_id=chain_id,
             token_network_registry_address=token_network_registry_address_test_default,
             pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-            matrix_servers=["matrix.example.com"],
         )
     assert config.url in urls
     assert is_canonical_address(config.payment_address)
@@ -105,7 +106,6 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
             node_network_id=chain_id,
             token_network_registry_address=token_network_registry_address_test_default,
             pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-            matrix_servers=["matrix.example.com"],
         )
     assert config.url == given_address
     assert is_same_address(config.payment_address, json_data["payment_address"])
@@ -123,7 +123,6 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
                 node_network_id=chain_id,
                 token_network_registry_address=token_network_registry_address_test_default,
                 pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-                matrix_servers=["http://matrix.example.com"],
             )
 
     # Addresses of token network registries of pfs and client conflict, should exit the client
@@ -139,7 +138,6 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
                     to_canonical_address("0x2222222222222222222222222222222222222221")
                 ),
                 pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-                matrix_servers=["http://matrix.example.com"],
             )
 
     # ChainIDs of pfs and client conflict, should exit the client
@@ -153,22 +151,59 @@ def test_configure_pfs(service_registry_address, private_keys, web3, contract_ma
                 node_network_id=ChainID(chain_id + 1),
                 token_network_registry_address=token_network_registry_address_test_default,
                 pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-                matrix_servers=["http://matrix.example.com"],
             )
 
-    # Wrong matrix server
-    response = mocked_json_response(response_data=json_data)
-    with pytest.raises(RaidenError, match="matrix server"):
-        with patch.object(session, "get", return_value=response):
-            configure_pfs_or_exit(
-                pfs_url="http://foo",
-                routing_mode=RoutingMode.PFS,
-                service_registry=service_registry,
-                node_network_id=ChainID(chain_id),
-                token_network_registry_address=token_network_registry_address_test_default,
-                pathfinding_max_fee=DEFAULT_PATHFINDING_MAX_FEE,
-                matrix_servers=["matrix.doesnotexist.com"],
-            )
+
+def test_check_pfs_transport_configuration(chain_id, private_keys, caplog):
+    matrix_server_url = "http://matrix.example.com"
+    matrix_room_id = "!room-id:matrix.example.com"
+    pfs_info = PFSInfo(
+        url="http://foo",
+        price=TokenAmount(0),
+        chain_id=chain_id,
+        token_network_registry_address=token_network_registry_address_test_default,
+        payment_address=to_canonical_address("0x2222222222222222222222222222222222222221"),
+        message="",
+        operator="",
+        version="",
+        user_deposit_address=privatekey_to_address(private_keys[1]),
+        confirmed_block_number=BlockNumber(10),
+        matrix_server=matrix_server_url,
+        matrix_room_id=matrix_room_id,
+    )
+
+    # Room id mismatch, must raise
+    with pytest.raises(RaidenError):
+        check_pfs_transport_configuration(
+            pfs_info=pfs_info,
+            pfs_was_autoselected=True,
+            transport_pfs_broadcast_room_id="!this-is-not-the-room-youre-looking-for:example.com",
+            matrix_server_url=matrix_server_url,
+            matrix_server_was_autoselected=True,
+        )
+
+    # Room ids match, must not raise
+    check_pfs_transport_configuration(
+        pfs_info=pfs_info,
+        pfs_was_autoselected=True,
+        transport_pfs_broadcast_room_id=matrix_room_id,
+        matrix_server_url=matrix_server_url,
+        matrix_server_was_autoselected=True,
+    )
+
+    # With the matrix_room_id missing from the PFS response the check can't be performed
+    pfs_info_no_room_id = dataclasses.replace(pfs_info, matrix_room_id=None)
+    with caplog.at_level(logging.WARNING):
+        check_pfs_transport_configuration(
+            pfs_info=pfs_info_no_room_id,
+            pfs_was_autoselected=True,
+            transport_pfs_broadcast_room_id="!not-this-again:matrix.org",
+            matrix_server_url=matrix_server_url,
+            matrix_server_was_autoselected=True,
+        )
+        assert "Can't check PFS transport configuration" in (
+            record.msg["event"] for record in caplog.records
+        )
 
 
 def test_check_pfs_for_production(
@@ -195,6 +230,7 @@ def test_check_pfs_for_production(
         user_deposit_address=privatekey_to_address(private_keys[1]),
         confirmed_block_number=BlockNumber(10),
         matrix_server="http://matrix.example.com",
+        matrix_room_id="!room-id:matrix.example.com",
     )
     with pytest.raises(RaidenError):
         check_pfs_for_production(service_registry=service_registry, pfs_info=pfs_info)
@@ -212,6 +248,7 @@ def test_check_pfs_for_production(
         user_deposit_address=privatekey_to_address(private_keys[1]),
         confirmed_block_number=BlockNumber(10),
         matrix_server="http://matrix.example.com",
+        matrix_room_id="!room-id:matrix.example.com",
     )
     with pytest.raises(RaidenError):
         check_pfs_for_production(service_registry=service_registry, pfs_info=pfs_info)

raiden/tests/integration/network/transport/test_matrix_transport.py

@@ -426,7 +426,7 @@ def test_join_invalid_discovery(
     transport.start(raiden_service, [], None)
     transport.log = MagicMock()
     discovery_room_name = make_room_alias(transport.chain_id, "discovery")
-    assert isinstance(transport._broadcast_rooms.get(discovery_room_name), Room)
+    assert isinstance(transport.broadcast_rooms.get(discovery_room_name), Room)
 
     transport.stop()
     transport.greenlet.get()
@@ -492,7 +492,7 @@ def test_matrix_discovery_room_offline_server(
 
     discovery_room_name = make_room_alias(transport.chain_id, "discovery")
     with gevent.Timeout(1):
-        while not isinstance(transport._broadcast_rooms.get(discovery_room_name), Room):
+        while not isinstance(transport.broadcast_rooms.get(discovery_room_name), Room):
             gevent.sleep(0.1)
 
     transport.stop()
@@ -524,7 +524,7 @@ def test_matrix_broadcast(
     gevent.idle()
 
     ms_room_name = make_room_alias(transport.chain_id, MONITORING_BROADCASTING_ROOM)
-    ms_room = transport._broadcast_rooms.get(ms_room_name)
+    ms_room = transport.broadcast_rooms.get(ms_room_name)
     assert isinstance(ms_room, Room)
 
     ms_room.send_text = MagicMock(spec=ms_room.send_text)
@@ -585,7 +585,7 @@ def test_monitoring_broadcast_messages(
     transport.start(raiden_service, [], None)
 
     ms_room_name = make_room_alias(transport.chain_id, MONITORING_BROADCASTING_ROOM)
-    ms_room = transport._broadcast_rooms.get(ms_room_name)
+    ms_room = transport.broadcast_rooms.get(ms_room_name)
     assert isinstance(ms_room, Room)
     ms_room.send_text = MagicMock(spec=ms_room.send_text)
 
@@ -666,7 +666,7 @@ def test_monitoring_broadcast_messages_in_production_if_bigger_than_threshold(
     transport.start(raiden_service, [], None)
 
     ms_room_name = make_room_alias(transport.chain_id, MONITORING_BROADCASTING_ROOM)
-    ms_room = transport._broadcast_rooms.get(ms_room_name)
+    ms_room = transport.broadcast_rooms.get(ms_room_name)
     assert isinstance(ms_room, Room)
     ms_room.send_text = MagicMock(spec=ms_room.send_text)
 
@@ -757,7 +757,7 @@ def test_pfs_broadcast_messages(
     transport.start(raiden_service, [], None)
 
     pfs_room_name = make_room_alias(transport.chain_id, PATH_FINDING_BROADCASTING_ROOM)
-    pfs_room = transport._broadcast_rooms.get(pfs_room_name)
+    pfs_room = transport.broadcast_rooms.get(pfs_room_name)
     assert isinstance(pfs_room, Room)
     pfs_room.send_text = MagicMock(spec=pfs_room.send_text)
 
@@ -1219,7 +1219,7 @@ def _handle_responses(
     transport2.start(raiden_service2, [], None)
 
     pfs_broadcast_room_alias = make_room_alias(transport0.chain_id, PATH_FINDING_BROADCASTING_ROOM)
-    pfs_broadcast_room_t0 = transport0._broadcast_rooms[pfs_broadcast_room_alias]
+    pfs_broadcast_room_t0 = transport0.broadcast_rooms[pfs_broadcast_room_alias]
 
     # Get the sync helper to control flow of asynchronous syncs
     sync_progress1 = transport1._client.sync_progress