Backport Capabilities for release

This ports protocol handshake (#3283 / #6503) capabilities from #6482 / #6572
back to the release branch.

Author: konradkonrad

raiden/constants.py

@@ -82,6 +82,15 @@ class Networks(Enum):
     SMOKETEST = ChainID(627)
 
 
+class Capabilities(Enum):
+    """Capabilities allow for protocol handshake between nodes."""
+
+    RECEIVE = "Receive"  # handle receiving transfers
+    MEDIATE = "Mediate"  # support for mediating transfers; mediating requires receiving
+    DELIVERY = "Delivery"  # expects and sends Delivery messages
+    WEBRTC = "webRTC"  # supports webRTC messaging
+
+
 # Set at 64 since parity's default is 64 and Geth's default is 128
 # TODO: Make this configurable. Since in parity this is also a configurable value
 STATE_PRUNING_AFTER_BLOCKS = 64

raiden/network/transport/matrix/transport.py

@@ -54,6 +54,7 @@
 from raiden.transfer.identifiers import CANONICAL_IDENTIFIER_UNORDERED_QUEUE, QueueIdentifier
 from raiden.transfer.state import NetworkState, QueueIdsToQueues
 from raiden.transfer.state_change import ActionChangeNodeNetworkState
+from raiden.utils.capabilities import capconfig_to_dict
 from raiden.utils.formatting import to_checksum_address, to_hex_address
 from raiden.utils.logging import redact_secret
 from raiden.utils.notifying_queue import NotifyingQueue
@@ -72,6 +73,7 @@
     MessageID,
     NamedTuple,
     Optional,
+    PeerCapabilities,
     RoomID,
     Set,
     Tuple,
@@ -443,10 +445,12 @@ def start(  # type: ignore
         self._address_mgr.start()
 
         try:
+            capabilities = capconfig_to_dict(self._config.capabilities_config)
             login(
                 client=self._client,
                 signer=self._raiden_service.signer,
                 prev_auth_data=prev_auth_data,
+                capabilities=capabilities,
             )
         except ValueError:
             # `ValueError` may be raised if `get_user` provides invalid data to
@@ -1377,7 +1381,7 @@ def _user_presence_changed(self, user: User, _presence: UserPresence) -> None:
         )
 
     def _address_reachability_changed(
-        self, address: Address, reachability: AddressReachability
+        self, address: Address, reachability: AddressReachability, capabilities: PeerCapabilities
     ) -> None:
         if reachability is AddressReachability.REACHABLE:
             node_reachability = NetworkState.REACHABLE
@@ -1392,6 +1396,8 @@ def _address_reachability_changed(
         else:
             raise TypeError(f'Unexpected reachability state "{reachability}".')
 
+        log.debug("Peer capabilities", capabilities=capabilities, address=address)
+
         assert self._raiden_service is not None, "_raiden_service not set"
         state_change = ActionChangeNodeNetworkState(address, node_reachability)
         self._raiden_service.handle_and_track_state_changes([state_change])

raiden/network/transport/matrix/utils.py

@@ -56,9 +56,10 @@
 )
 from raiden.network.utils import get_average_http_response_time
 from raiden.storage.serialization.serializer import MessageSerializer
+from raiden.utils.capabilities import deserialize_capabilities, serialize_capabilities
 from raiden.utils.gevent import spawn_named
 from raiden.utils.signer import Signer, recover
-from raiden.utils.typing import Address, ChainID, MessageID, Signature
+from raiden.utils.typing import Address, ChainID, MessageID, PeerCapabilities, Signature
 from raiden_contracts.constants import ID_TO_CHAINNAME
 
 log = structlog.get_logger(__name__)
@@ -186,7 +187,9 @@ def __init__(
         self,
         client: GMatrixClient,
         displayname_cache: DisplayNameCache,
-        address_reachability_changed_callback: Callable[[Address, AddressReachability], None],
+        address_reachability_changed_callback: Callable[
+            [Address, AddressReachability, PeerCapabilities], None
+        ],
         user_presence_changed_callback: Optional[Callable[[User, UserPresence], None]] = None,
         _log_context: Optional[Dict[str, Any]] = None,
     ) -> None:
@@ -269,6 +272,10 @@ def get_address_reachability_state(self, address: Address) -> ReachabilityState:
         """ Return the current reachability state for ``address``. """
         return self._address_to_reachabilitystate.get(address, UNKNOWN_REACHABILITY_STATE)
 
+    def get_address_capabilities(self, address: Address) -> PeerCapabilities:
+        """ Return the protocol capabilities for ``address``. """
+        return self._address_to_capabilities.get(address, PeerCapabilities({}))
+
     def force_user_presence(self, user: User, presence: UserPresence) -> None:
         """ Forcibly set the ``user`` presence to ``presence``.
 
@@ -331,6 +338,18 @@ def track_address_presence(
 
         self._maybe_address_reachability_changed(address)
 
+    def query_capabilities_for_user_id(self, user_id: str) -> PeerCapabilities:
+        """ This pulls the `avatar_url` for a given user/user_id and parses the capabilities.  """
+        try:
+            user: User = self._client.get_user(user_id)
+        except MatrixRequestError:
+            return PeerCapabilities({})
+        avatar_url = user.get_avatar_url()
+        if avatar_url is not None:
+            return PeerCapabilities(deserialize_capabilities(avatar_url))
+        else:
+            return PeerCapabilities({})
+
     def get_reachability_from_matrix(self, user_ids: Iterable[str]) -> AddressReachability:
         """ Get the current reachability without any side effects
 
@@ -347,11 +366,14 @@ def get_reachability_from_matrix(self, user_ids: Iterable[str]) -> AddressReacha
     def _maybe_address_reachability_changed(self, address: Address) -> None:
         # A Raiden node may have multiple Matrix users, this happens when
         # Raiden roams from a Matrix server to another. This loop goes over all
-        # these users and uses the "best" presence. IOW, if there is a single
+        # these users and uses the "best" presence. IOW, if there is at least one
         # Matrix user that is reachable, then the Raiden node is considered
         # reachable.
         userids = self._address_to_userids[address].copy()
-        composite_presence = {self._userid_to_presence.get(uid) for uid in userids}
+        presence_to_uid = defaultdict(list)
+        for uid in userids:
+            presence_to_uid[self._userid_to_presence.get(uid)].append(uid)
+        composite_presence = set(presence_to_uid.keys())
 
         new_presence = UserPresence.UNKNOWN
         for presence in UserPresence.__members__.values():
@@ -364,7 +386,9 @@ def _maybe_address_reachability_changed(self, address: Address) -> None:
         prev_reachability_state = self.get_address_reachability_state(address)
         if new_address_reachability == prev_reachability_state.reachability:
             return
-
+        # for capabilities, we get the "first" uid that showed the `new_presence`
+        present_uid = presence_to_uid[new_presence].pop()
+        capabilities = self.query_capabilities_for_user_id(present_uid)
         now = datetime.now()
 
         self.log.debug(
@@ -379,8 +403,11 @@ def _maybe_address_reachability_changed(self, address: Address) -> None:
         self._address_to_reachabilitystate[address] = ReachabilityState(
             new_address_reachability, now
         )
+        self._address_to_capabilities[address] = capabilities
 
-        self._address_reachability_changed_callback(address, new_address_reachability)
+        self._address_reachability_changed_callback(
+            address, new_address_reachability, capabilities
+        )
 
     def _presence_listener(self, event: Dict[str, Any], presence_update_id: int) -> None:
         """
@@ -437,6 +464,7 @@ def _presence_listener(self, event: Dict[str, Any], presence_update_id: int) ->
     def _reset_state(self) -> None:
         self._address_to_userids: Dict[Address, Set[str]] = defaultdict(set)
         self._address_to_reachabilitystate: Dict[Address, ReachabilityState] = dict()
+        self._address_to_capabilities: Dict[Address, PeerCapabilities] = dict()
         self._userid_to_presence: Dict[str, UserPresence] = dict()
         self._userid_to_presence_update_id: Dict[str, int] = dict()
 
@@ -569,7 +597,7 @@ def join_broadcast_room(client: GMatrixClient, broadcast_room_alias: str) -> Roo
         )
 
 
-def first_login(client: GMatrixClient, signer: Signer, username: str) -> User:
+def first_login(client: GMatrixClient, signer: Signer, username: str, cap_str: str) -> User:
     """Login within a server.
 
     There are multiple cases where a previous auth token can become invalid and
@@ -635,6 +663,12 @@ def first_login(client: GMatrixClient, signer: Signer, username: str) -> User:
     if current_display_name != signature_hex:
         user.set_display_name(signature_hex)
 
+    current_capabilities = user.get_avatar_url() or ""
+
+    # Only set the capabilities if necessary.
+    if current_capabilities != cap_str:
+        user.set_avatar_url(cap_str)
+
     log.debug(
         "Logged in",
         node=to_checksum_address(username),
@@ -679,14 +713,21 @@ def login_with_token(client: GMatrixClient, user_id: str, access_token: str) ->
     return client.get_user(client.user_id)
 
 
-def login(client: GMatrixClient, signer: Signer, prev_auth_data: Optional[str] = None) -> User:
-    """ Login with a matrix server.
+def login(
+    client: GMatrixClient,
+    signer: Signer,
+    prev_auth_data: Optional[str] = None,
+    capabilities: Dict[str, Any] = None,
+) -> User:
+    """Login with a matrix server.
 
     Params:
         client: GMatrixClient instance configured with desired homeserver.
         signer: Signer used to sign the password and displayname.
         prev_auth_data: Previously persisted authentication using the format "{user}/{password}".
     """
+    if capabilities is None:
+        capabilities = {}
     server_url = client.api.base_url
     server_name = urlparse(server_url).netloc
 
@@ -708,7 +749,11 @@ def login(client: GMatrixClient, signer: Signer, prev_auth_data: Optional[str] =
                 server_name=server_name,
             )
 
-    return first_login(client, signer, username)
+    try:
+        capstr = serialize_capabilities(capabilities)
+    except ValueError:
+        raise Exception("error serializing")
+    return first_login(client, signer, username, capstr)
 
 
 @cached(cache=LRUCache(128), key=attrgetter("user_id", "displayname"), lock=Semaphore())

raiden/settings.py

@@ -127,6 +127,14 @@ def get_proportional_imbalance_fee(self, token_address: TokenAddress) -> Proport
         )
 
 
+@dataclass
+class CapabilitiesConfig:
+    receive: bool = True
+    mediate: bool = True
+    delivery: bool = True
+    web_rtc: bool = False
+
+
 @dataclass
 class MatrixTransportConfig:
     retries_before_backoff: int
@@ -137,6 +145,7 @@ class MatrixTransportConfig:
     available_servers: List[str]
     sync_timeout: int = DEFAULT_TRANSPORT_MATRIX_SYNC_TIMEOUT
     sync_latency: int = DEFAULT_TRANSPORT_MATRIX_SYNC_LATENCY
+    capabilities_config: CapabilitiesConfig = CapabilitiesConfig()
 
 
 @dataclass
@@ -201,6 +210,7 @@ class RaidenConfig:
         retry_interval_max=DEFAULT_TRANSPORT_MATRIX_RETRY_INTERVAL_MAX,
         server=MATRIX_AUTO_SELECT_SERVER,
         sync_timeout=DEFAULT_TRANSPORT_MATRIX_SYNC_TIMEOUT,
+        capabilities_config=CapabilitiesConfig(),
     )
 
     rest_api: RestApiConfig = RestApiConfig()

raiden/tests/integration/fixtures/raiden_network.py

@@ -8,6 +8,7 @@
 
 from raiden.app import App
 from raiden.constants import Environment, RoutingMode
+from raiden.settings import CapabilitiesConfig
 from raiden.tests.utils.network import (
     CHAIN,
     BlockchainServices,
@@ -82,6 +83,7 @@ def raiden_chain(
     resolver_ports: List[Optional[int]],
     enable_rest_api: bool,
     port_generator: Iterator[Port],
+    capabilities: CapabilitiesConfig,
 ) -> Iterable[List[App]]:
 
     if len(token_addresses) != 1:
@@ -122,6 +124,7 @@ def raiden_chain(
         resolver_ports=resolver_ports,
         enable_rest_api=enable_rest_api,
         port_generator=port_generator,
+        capabilities_config=capabilities,
     )
 
     confirmed_block = raiden_apps[0].raiden.confirmation_blocks + 1
@@ -192,6 +195,19 @@ def resolvers(resolver_ports):
             resolver.terminate()
 
 
+@pytest.fixture
+def adhoc_capability():
+    return False
+
+
+@pytest.fixture
+def capabilities(adhoc_capability) -> CapabilitiesConfig:
+    config = CapabilitiesConfig()
+    if adhoc_capability:
+        config.adhoc_capability = adhoc_capability  # type: ignore
+    return config
+
+
 @pytest.fixture
 def raiden_network(
     token_addresses: List[TokenAddress],
@@ -222,6 +238,7 @@ def raiden_network(
     resolver_ports: List[Optional[int]],
     enable_rest_api: bool,
     port_generator: Iterator[Port],
+    capabilities: CapabilitiesConfig,
 ) -> Iterable[List[App]]:
     service_registry_address = None
     if blockchain_services.service_registry:
@@ -254,6 +271,7 @@ def raiden_network(
         resolver_ports=resolver_ports,
         enable_rest_api=enable_rest_api,
         port_generator=port_generator,
+        capabilities_config=capabilities,
     )
 
     confirmed_block = raiden_apps[0].raiden.confirmation_blocks + 1

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

@@ -57,8 +57,9 @@
 from raiden.transfer.identifiers import CANONICAL_IDENTIFIER_UNORDERED_QUEUE, QueueIdentifier
 from raiden.transfer.state import NetworkState
 from raiden.transfer.state_change import ActionChannelClose
+from raiden.utils.capabilities import capconfig_to_dict, deserialize_capabilities
 from raiden.utils.formatting import to_checksum_address
-from raiden.utils.typing import Address, Dict, List, TokenNetworkAddress, cast
+from raiden.utils.typing import Address, Dict, List, PeerCapabilities, TokenNetworkAddress, cast
 from raiden.waiting import wait_for_network_state
 
 HOP1_BALANCE_PROOF = factories.BalanceProofSignedStateProperties(pkey=factories.HOP1_KEY)
@@ -1333,3 +1334,40 @@ def test_transport_presence_updates(raiden_network, restart_node, retry_timeout)
     app2.raiden.transport.immediate_health_check_for(app1.raiden.address)
     wait_for_network_state(app0.raiden, app2.raiden.address, NetworkState.REACHABLE, retry_timeout)
     wait_for_network_state(app1.raiden, app2.raiden.address, NetworkState.REACHABLE, retry_timeout)
+
+
+@raise_on_failure
+@pytest.mark.parametrize("matrix_server_count", [1])
+@pytest.mark.parametrize("number_of_nodes", [2])
+@pytest.mark.parametrize("adhoc_capability", [True])
+@pytest.mark.parametrize(
+    "broadcast_rooms", [[DISCOVERY_DEFAULT_ROOM, PATH_FINDING_BROADCASTING_ROOM]]
+)
+def test_transport_capabilities(raiden_network: List[App], capabilities, retry_timeout):
+    """
+    Test that raiden matrix users have the `avatar_url` set in a format understood
+    by the capabilities parser.
+    """
+    app0_outer, app1_outer = raiden_network
+    app0 = app0_outer.raiden
+    app1 = app1_outer.raiden
+
+    app0.transport.immediate_health_check_for(app1.address)
+    app1.transport.immediate_health_check_for(app0.address)
+
+    wait_for_network_state(app0, app1.address, NetworkState.REACHABLE, retry_timeout)
+    wait_for_network_state(app1, app0.address, NetworkState.REACHABLE, retry_timeout)
+
+    expected_capabilities = capconfig_to_dict(capabilities)
+
+    app1_user_ids = app0.transport.get_user_ids_for_address(app1.address)
+    assert len(app1_user_ids) == 1, "app1 should have exactly one user_id"
+    app1_user = app0.transport._client.get_user(app1_user_ids.pop())
+    app1_avatar_url = app1_user.get_avatar_url()
+    assert len(app1_avatar_url), "avatar_url not set for app1"
+    app1_capabilities = deserialize_capabilities(app1_avatar_url)
+    assert "adhoc_capability" in app1_capabilities, "capabilities could not be parsed correctly"
+
+    msg = "capabilities were not collected in transport client"
+    collected_capabilities = app0.transport._address_mgr.get_address_capabilities(app1.address)
+    assert collected_capabilities == PeerCapabilities(expected_capabilities), msg