new: Choose exits that implement congestion control

and create methods to check that the consensus implements congestion
control.

Closes: #40125.

INSTALL.rst

@@ -23,6 +23,7 @@ System requirements
 --------------------
 
 - Tor (last stable version is recommended)
+  To use only exits that implement congestion control, tor >= 0.4.7.4-alpha-dev
 - Python 3 (>= 3.7)
 
 Python dependencies

docs/source/activity_second_relay.puml

@@ -2,13 +2,29 @@
 
 start
 
-if (relay to measure is exit?) then (yes)
-  :obtain non-exits;
+if (consensus has `cc_alg=2`) then (yes)
+  if (consensus has `bwscanner_cc>=1`) then (yes)
+    :obtain exits
+    with proto `FlowCtrol=2`
+    without bad flag
+    that can exit
+    to port 443;
+  else (no)
+    :obtain exits
+    with proto `FlowCtrol!=2`
+    without bad flag
+    that can exit
+    to port 443;
+  endif
 else (no)
-  :obtain an exits
-  without bad flag
-  that can exit
-  to port 443;
+  if (relay to measure is exit?) then (yes)
+    :obtain non-exits;
+  else (no)
+    :obtain an exits
+    without bad flag
+    that can exit
+    to port 443;
+  endif
 endif
 :potential second relays;
 :obtain a relay
@@ -27,4 +43,4 @@ whith exit as
 second hop;
 stop
 
-@enduml
\ No newline at end of file
+@enduml

docs/source/how_works.rst

@@ -85,6 +85,8 @@ Source code: :func:`sbws.core.scanner.measure_relay`
 Selecting a second relay
 ------------------------
 
+#. If the consensus has `cc_alg=2` param, use the exits that have `2` in the
+   field `FlowCtrl` in their `proto` line, otherwise
 #. If the relay to measure is an exit, use it as an exit and obtain the
    non-exits.
 #. If the relay to measure is not an exit, use it as first hop and obtain

sbws/core/scanner.py

@@ -223,10 +223,27 @@ def _pick_ideal_second_hop(relay, dest, rl, cont, is_exit):
     # In the case that a concrete exit can't exit to the Web server, it is not
     # a problem since the relay will be measured in the next loop with other
     # random exit.
-    candidates = (
-        rl.exits_not_bad_allowing_port(dest.port) if is_exit else rl.non_exits
-    )
+
+    # #40125
+    if rl.is_consensus_cc_alg_2:
+        if rl.is_consensus_bwscanner_cc_gte_1:
+            log.debug("Congestion control enabled.")
+            candidates = rl.exits_with_2_in_flowctrl(dest.port)
+        else:  # bwscanner_cc != 1
+            log.debug(
+                "Congestion control enabled but not using exits with "
+                "congestion control enabled."
+            )
+            candidates = rl.exits_without_2_in_flowctrl(dest.port)
+    else:
+        log.debug("Congestion control disabled.")
+        candidates = (
+            rl.exits_not_bad_allowing_port(dest.port)
+            if is_exit
+            else rl.non_exits
+        )
     if not len(candidates):
+        log.debug("No candidates.")
         return None
     # In the case the helper is an exit, the entry could be an exit too
     # (#40041), so ensure the helper is not the same as the entry, likely to
@@ -235,6 +252,8 @@ def _pick_ideal_second_hop(relay, dest, rl, cont, is_exit):
         candidates = [
             c for c in candidates if c.fingerprint != relay.fingerprint
         ]
+    # While not all exits implement congestion control, the min bw might not
+    # correspond to the subset that implement it.
     min_relay_bw = rl.exit_min_bw() if is_exit else rl.non_exit_min_bw()
     log.debug(
         "Picking a 2nd hop to measure %s from %d choices. is_exit=%s",
@@ -295,9 +314,11 @@ def create_path_relay(relay, dest, rl, cb, relay_as_entry=True):
     # is True when the relay is the entry (helper has to be exit)
     # and False when the relay is not the entry, ie. is the exit (helper does
     # not have to be an exit)
+
     helper = _pick_ideal_second_hop(
         relay, dest, rl, cb.controller, is_exit=relay_as_entry
     )
+
     if not helper:
         return error_no_helper(relay, dest)
     if relay_as_entry:

sbws/globals.py

@@ -61,7 +61,7 @@ TORRC_OPTIONS_CAN_FAIL = OrderedDict(
     {
         # Since currently scanner anonymity is not the goal, ConnectionPadding
         # is disable to do not send extra traffic
-        "ConnectionPadding": "0"
+        "ConnectionPadding": "0",
     }
 )
 

sbws/lib/relaylist.py

@@ -125,6 +125,23 @@ class Relay:
     def observed_bandwidth(self):
         return self._from_desc("observed_bandwidth")
 
+    @property
+    def has_2_in_flowctrl(self):
+        """
+        Return True if the `FlowCtrl` field in the relay's protover consensus
+        line include a value of 2 and False otherwise.
+
+        """
+        # NOTE: stem doesn't seem to obtain `pr`` nor create `protocols``
+        # protocols = self._from_ns("protocols")
+        protocols = self._from_desc("protocols")
+        if protocols:
+            if 2 in protocols.get("FlowCtrl", []):
+                log.debug("Exit %s has 2 in `FlowCtrl`.", self.nickname)
+                return True
+        log.debug("Exit %s does not have 2 in `FlowCtrl`.", self.nickname)
+        return False
+
     @property
     def consensus_bandwidth(self):
         """Return the consensus bandwidth in Bytes.
@@ -484,12 +501,121 @@ class RelayList:
         # Calculate minimum bandwidth value for 2nd hop after we refreshed
         # our available relays.
         self._calculate_min_bw_second_hop()
+        self.set_consensus_params()
 
     @property
     def recent_consensus_count(self):
         """Number of times a new consensus was obtained."""
         return len(self._recent_consensus)
 
+    def exits_with_2_in_flowctrl(self, port):
+        """
+        Return the exits that include a value of 2 in the `FlowCtrl` field
+        in their protover consensus line.
+
+        """
+        return [
+            r
+            for r in self.exits_not_bad_allowing_port(port)
+            if r.has_2_in_flowctrl
+        ]
+
+    def exits_without_2_in_flowctrl(self, port):
+        """
+        Return the exits that do not include a value of 2 in the `FlowCtrl`
+        field in their protover consensus line or the field is missing.
+
+        """
+        return [
+            r
+            for r in self.exits_not_bad_allowing_port(port)
+            if not r.has_2_in_flowctrl
+        ]
+
+    def set_consensus_params(self):
+        """Obtain current consensus params fields and store them as an attr.
+
+        It is not possible to obtain them from `get_network_statuses` via
+        control port, only via a cached file.
+
+        """
+        if self._controller.get_conf("TestingTorNetwork") == "1":
+            log.debug("In a testing network.")
+            self.consensus_params_dict = {}
+            return
+        log.debug("Not in a testing network.")
+        consensus = self._controller.get_info(
+            "dir/status-vote/current/consensus"
+        )
+        from unittest import mock
+
+        if isinstance(consensus, mock.Mock):
+            log.debug("Mocked consensus.")
+            self.consensus_params_dict = {}
+            return
+        # Create a dictionary from all the consensus lines.
+        consensus_dict = dict(
+            [
+                (line.split(" ")[0], line.split()[1:])
+                for line in consensus.split("\n")
+            ]
+        )
+        # Create a dictionary from the consensus `params` line.
+        self.consensus_params_dict = dict(
+            [
+                (p.split("=")[0], p.split("=")[1])
+                for p in consensus_dict.get("params", [])
+            ]
+        )
+        log.debug("Consensus params: %s", self.consensus_params_dict.keys())
+
+    @property
+    def is_consensus_cc_alg_2(self):
+        """
+        Return True if the consensus document has a value of 2 in the `cc_alg`
+        field.
+
+        From proposals/324-rtt-congestion-control.txt spec::
+
+            6.5.1. Parameters common to all algorithms
+            [...]
+            cc_alg:
+            - Description:
+                Specifies which congestion control algorithm clients should
+                use, as an integer.
+            - Range: [0,3]  (0=fixed, 1=Westwood, 2=Vegas, 3=NOLA)
+            - Default: 2
+
+        """
+        if (
+            self.consensus_params_dict
+            and self.consensus_params_dict.get("cc_alg", None) == 2
+        ):
+            log.info("The consensus implements congestion control.")
+            return True
+        log.info("The consensus does not implement congestion control.")
+        return False
+
+    @property
+    def is_consensus_bwscanner_cc_gte_1(self):
+        """
+        Return True if the consensus document has a value of 1 or greater in
+        the `bwscanner_cc` field."""
+        if (
+            self.consensus_params_dict
+            and self.consensus_params_dict.get("bwscanner_cc", None) >= 1
+        ):
+            log.info(
+                "The consensus says to use exits that support congestion"
+                " control."
+            )
+            return True
+        log.info(
+            "The consensus says to use exits that do not support congestion"
+            " control."
+        )
+        return False
+
     def exits_not_bad_allowing_port(self, port, strict=False):
         return [
             r

tests/conftest.py

@@ -79,6 +79,7 @@ def router_statuses_5days_later(root_data_path):
 @pytest.fixture(scope="session")
 def controller(router_statuses):
     controller = mock.Mock()
+    controller.get_info.return_value = "params foo=bar"
     controller.get_network_statuses.return_value = router_statuses
     return controller
 

tests/integration/conftest.py

@@ -3,6 +3,7 @@ import argparse
 import os
 
 import pytest
+from stem.version import Version
 
 from sbws.lib.circuitbuilder import GapsCircuitBuilder as CB
 from sbws.lib.destination import DestinationList
@@ -93,7 +94,8 @@ def persistent_launch_tor(conf):
 
 @pytest.fixture(scope="session")
 def rl(args, conf, persistent_launch_tor):
-    return RelayList(args, conf, persistent_launch_tor)
+    temp_rl = RelayList(args, conf, persistent_launch_tor)
+    return temp_rl
 
 
 @pytest.fixture(scope="session")
@@ -110,4 +112,8 @@ def dests(args, conf, persistent_launch_tor, cb, rl):
     return dests
 
 
-# @pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
+def is_cc_tor_version(persistent_launch_tor):
+    version = persistent_launch_tor.get_version()
+    print("version", version)
+    return version >= Version("0.4.7.4-alpha-dev")

tests/integration/core/test_scanner.py

@@ -2,7 +2,7 @@ import logging
 
 import pytest
 
-from sbws.core.scanner import measure_relay
+from sbws.core.scanner import _pick_ideal_second_hop, measure_relay
 from sbws.lib.resultdump import ResultSuccess
 
 
@@ -70,3 +70,22 @@ def test_measure_relay_with_relaybandwidthrate(
     dls = result.downloads
     for dl in dls:
         assert_within(dl["amount"] / dl["duration"], one_mbyte, allowed_error)
+
+
+def test_second_hop_has_2_in_flowctrl(
+    is_cc_tor_version, dests, rl, persistent_launch_tor
+):
+    if not is_cc_tor_version:
+        import pytest
+
+        pytest.skip("This test can't be run with this tor version")
+        return
+    rl.consensus_params_dict = {"cc_alg": 2, "bwscanner_cc": 1}
+    assert rl.is_consensus_cc_alg_2
+    assert rl.is_consensus_bwscanner_cc_gte_1
+    dest = dests._all_dests[0]
+    relay = rl._relays[0]
+    helper = _pick_ideal_second_hop(
+        relay, dest, rl, persistent_launch_tor, False
+    )
+    assert helper.has_2_in_flowctrl

tests/integration/lib/test_circuitbuilder.py

@@ -21,3 +21,28 @@ def test_build_circuit(cb, rl):
     path = [entry.fingerprint, exit_relay.fingerprint]
     circuit_id, _ = cb.build_circuit(path)
     assert circuit_id
+
+
+def test_build_circuit_with_exit_flowctrl(is_cc_tor_version, cb, rl):
+    if not is_cc_tor_version:
+        import pytest
+
+        pytest.skip("This test can't be run with this tor version")
+        return
+    rl.consensus_params_dict = {"cc_alg": 2, "bwscanner_cc": 1}
+    # Path is empty
+    path = []
+    circuit_id, _ = cb.build_circuit(path)
+    assert not circuit_id
+    # Valid path, not valid exit
+    exits = rl.exits_with_2_in_flowctrl(port=443)
+    # See https://gitlab.torproject.org/tpo/core/chutney/-/issues/40013:
+    # Work around to get supposed non-exits because chutney is putting Exit
+    # flag to all relays
+    non_exits = list(set(rl.exits).difference(set(exits)))
+    entry = random.choice(non_exits)
+    # Valid path and relays
+    exit_relay = random.choice(exits)
+    path = [entry.fingerprint, exit_relay.fingerprint]
+    circuit_id, _ = cb.build_circuit(path)
+    assert circuit_id