Logic error in FTS3 transfertool checksum validation strategy
Motivation
----------
The function `checksum_validation_strategy` in `rucio/transfertool/fts3.py` determines which checksumming strategy to apply to a transfer. This function has some errors, for instance comparing the variables `source_supported_checksums` and `dest_supported_checksums` with `['none']`, instead of the correct value `[]`.

Another error is when the source supports checksumming, but the destination does not, which implies there are no common checksum algorithms. In this case case the checksum validation strategy is still selected as `destination`, and not `source`, despite the fact that the destination doesn't have any supported algorithms. This causes checksumming to be skipped on the source, even when it should have been performed.

Relevant source for context:

```python3
def checksum_validation_strategy(src_attributes, dst_attributes, logger):
    source_supported_checksums = get_rse_supported_checksums_from_attributes(src_attributes)
    dest_supported_checksums = get_rse_supported_checksums_from_attributes(dst_attributes)
    common_checksum_names = set(source_supported_checksums).intersection(dest_supported_checksums)

    ...

    if len(common_checksum_names) == 0:
        logger(logging.INFO, 'No common checksum method. Verifying destination only.')
        verify_checksum = 'destination'

    ...
```

Modification
------------
Determine the correct expected behaviour and rewrite `checksum_validation_strategy`.


diff --git a/lib/rucio/common/utils.py b/lib/rucio/common/utils.py
index ecbf2f2c8..0e097d876 100644
--- a/lib/rucio/common/utils.py
+++ b/lib/rucio/common/utils.py
@@ -228,14 +228,6 @@ def set_preferred_checksum(checksum_name):
         PREFERRED_CHECKSUM = checksum_name
 
 
-def set_checksum_value(file, checksum_names_list):
-    for checksum_name in checksum_names_list:
-        if checksum_name in file['metadata'].keys() and file['metadata'][checksum_name]:
-            file['checksum'] = '%s:%s' % (checksum_name.upper(), str(file['metadata'][checksum_name]))
-            if checksum_name == PREFERRED_CHECKSUM:
-                break
-
-
 def adler32(file):
     """
     An Adler-32 checksum is obtained by calculating two 16-bit checksums A and B
diff --git a/lib/rucio/core/rse.py b/lib/rucio/core/rse.py
index 77e3b6100..3945de959 100644
--- a/lib/rucio/core/rse.py
+++ b/lib/rucio/core/rse.py
@@ -32,7 +32,7 @@ import rucio.core.account_counter
 from rucio.common import exception, utils
 from rucio.common.cache import make_region_memcached
 from rucio.common.config import get_lfn2pfn_algorithm_default
-from rucio.common.utils import CHECKSUM_KEY, Availability, is_checksum_valid, GLOBALLY_SUPPORTED_CHECKSUMS
+from rucio.common.utils import CHECKSUM_KEY, GLOBALLY_SUPPORTED_CHECKSUMS, Availability
 from rucio.core.rse_counter import add_counter, get_counter
 from rucio.db.sqla import models
 from rucio.db.sqla.constants import (RSEType, ReplicaState)
@@ -710,14 +710,6 @@ def get_rse_attributes(rse_id, session=None):
     return result
 
 
-@read_session
-def get_rse_supported_checksums(rse_id, session=None):
-    """
-    Retrieve from the DB and parse the RSE attribute defining the checksum supported by the RSE
-    """
-    return parse_checksum_support_attribute(get_rse_attribute(key=CHECKSUM_KEY, rse_id=rse_id, session=session))
-
-
 def get_rse_supported_checksums_from_attributes(rse_attributes):
     """
     Parse the RSE attribute defining the checksum supported by the RSE
@@ -739,29 +731,12 @@ def parse_checksum_support_attribute(checksum_attribute):
     if not checksum_attribute:
         return GLOBALLY_SUPPORTED_CHECKSUMS
     else:
-        supported_checksum_list = checksum_attribute[0].split(',')
+        supported_checksum_list = [c.strip() for c in checksum_attribute.split(',') if c.strip()]
         if 'none' in supported_checksum_list:
             return []
         return supported_checksum_list
 
 
-@read_session
-def get_rse_is_checksum_supported(checksum_name, rse_id=None, session=None):
-    """
-    Retrieve RSE attribute value.
-
-    :param checksum_name: The desired checksum name for the attribute.
-    :param rse_id: The RSE id.
-    :param session: The database session in use.
-
-    :returns: True if required checksum is supported, False otherwise.
-    """
-    if is_checksum_valid(checksum_name):
-        return checksum_name in get_rse_supported_checksums(rse_id=rse_id, session=session)
-    else:
-        return False
-
-
 @transactional_session
 def set_rse_usage(rse_id, source, used, free, files=None, session=None):
     """
diff --git a/lib/rucio/core/transfer.py b/lib/rucio/core/transfer.py
index 5b9d2947f..6c74b6eed 100644
--- a/lib/rucio/core/transfer.py
+++ b/lib/rucio/core/transfer.py
@@ -137,8 +137,6 @@ class DirectTransferDefinition:
         self.operation_src = operation_src
         self.operation_dest = operation_dest
 
-        self.dict_attributes = {}
-
         self._dest_url = None
         self._legacy_sources = None
 
@@ -157,15 +155,6 @@ class DirectTransferDefinition:
     def dst(self):
         return self.destination
 
-    def __setitem__(self, key, value):
-        self.dict_attributes[key] = value
-
-    def __getitem__(self, key):
-        return self.dict_attributes[key]
-
-    def get(self, key, default=None):
-        return self.dict_attributes.get(key, default)
-
     @property
     def dest_url(self):
         if not self._dest_url:
diff --git a/lib/rucio/tests/test_conveyor.py b/lib/rucio/tests/test_conveyor.py
index 6dc5698cf..7be25eb6e 100644
--- a/lib/rucio/tests/test_conveyor.py
+++ b/lib/rucio/tests/test_conveyor.py
@@ -855,10 +855,9 @@ class FTSWrapper(FTS3Transfertool):
     def on_receive(job_response):
         pass
 
-    @classmethod
-    def _FTS3Transfertool__file_from_transfer(cls, transfer, job_params):
-        file = super()._FTS3Transfertool__file_from_transfer(transfer, job_params)
-        cls.on_submit(file)
+    def _FTS3Transfertool__file_from_transfer(self, transfer):
+        file = super()._FTS3Transfertool__file_from_transfer(transfer)
+        self.on_submit(file)
         return file
 
     def _FTS3Transfertool__bulk_query_responses(self, jobs_response, requests_by_eid):
@@ -1294,3 +1293,56 @@ def test_two_multihops_same_intermediate_rse(rse_factory, did_factory, root_acco
     for rse_id in [rse2_id, rse3_id, rse4_id, rse6_id]:
         with pytest.raises(ReplicaNotFound):
             replica_core.get_replica(rse_id=rse_id, **did)
+
+
+@skip_rse_tests_with_accounts
+@pytest.mark.noparallel(reason="runs submitter; poller and finisher")
+def test_checksum_validation(rse_factory, did_factory, root_account):
+    """
+    Ensure that the correct checksum validation strategy is applied on submission
+    """
+    src_rse, src_rse_id = rse_factory.make_rse(scheme='mock', protocol_impl='rucio.rse.protocols.posix.Default')
+    dst_rse1, dst_rse1_id = rse_factory.make_rse(scheme='mock', protocol_impl='rucio.rse.protocols.posix.Default')
+    dst_rse2, dst_rse2_id = rse_factory.make_rse(scheme='mock', protocol_impl='rucio.rse.protocols.posix.Default')
+    dst_rse3, dst_rse3_id = rse_factory.make_rse(scheme='mock', protocol_impl='rucio.rse.protocols.posix.Default')
+    all_rses = [src_rse_id, dst_rse1_id, dst_rse2_id, dst_rse3_id]
+
+    for rse_id in [dst_rse1_id, dst_rse2_id, dst_rse3_id]:
+        distance_core.add_distance(src_rse_id, rse_id, ranking=10)
+    for rse_id in all_rses:
+        rse_core.add_rse_attribute(rse_id, 'fts', TEST_FTS_HOST)
+
+    rse_core.add_rse_attribute(src_rse_id, 'supported_checksums', 'adler32')
+    rse_core.add_rse_attribute(dst_rse1_id, 'verify_checksum', False)
+    rse_core.add_rse_attribute(dst_rse2_id, 'supported_checksums', 'md5')
+    rse_core.add_rse_attribute(dst_rse3_id, 'supported_checksums', 'md5,adler32')
+
+    did = did_factory.upload_test_file(src_rse)
+    replica = replica_core.get_replica(rse_id=src_rse_id, **did)
+
+    rule_core.add_rule(dids=[did], account=root_account, copies=3, rse_expression=f'{dst_rse1}|{dst_rse2}|{dst_rse3}', grouping='ALL', weight=None, lifetime=None, locked=False, subscription_id=None)
+
+    class _FTSWrapper(FTSWrapper):
+        @staticmethod
+        def on_submit(file):
+            # Set the correct checksum on source and simulate a wrong checksum on destination
+            file['sources'] = [set_query_parameters(s_url, {'checksum': replica['adler32']}) for s_url in file['sources']]
+            file['destinations'] = [set_query_parameters(d_url, {'checksum': 'randomString2'}) for d_url in file['destinations']]
+
+    with patch('rucio.daemons.conveyor.submitter.TRANSFERTOOL_CLASSES_BY_NAME') as tt_mock:
+        tt_mock.__getitem__.return_value = _FTSWrapper
+        submitter(once=True, rses=[{'id': rse_id} for rse_id in all_rses], group_bulk=2, partition_wait_time=None, transfertype='single', filter_transfertool=None)
+
+    # Checksum verification disabled on this rse, so the transfer must use source validation and succeed
+    request = __wait_for_request_state(dst_rse_id=dst_rse1_id, state=RequestState.DONE, **did)
+    assert request['state'] == RequestState.DONE
+
+    # No common supported checksum between the source and destination rse. It will verify the destination rse checksum and fail
+    request = __wait_for_request_state(dst_rse_id=dst_rse2_id, state=RequestState.FAILED, **did)
+    assert request['state'] == RequestState.FAILED
+    assert 'User and destination checksums do not match' in request['err_msg']
+
+    # Common checksum exists between the two. It must use "both" validation strategy and fail
+    request = __wait_for_request_state(dst_rse_id=dst_rse3_id, state=RequestState.FAILED, **did)
+    assert 'Source and destination checksums do not match' in request['err_msg']
+    assert request['state'] == RequestState.FAILED
diff --git a/lib/rucio/transfertool/fts3.py b/lib/rucio/transfertool/fts3.py
index a9ab264ee..00a55cbc8 100644
--- a/lib/rucio/transfertool/fts3.py
+++ b/lib/rucio/transfertool/fts3.py
@@ -19,8 +19,9 @@ import functools
 import logging
 import time
 import traceback
-from urllib.parse import urlparse
 import uuid
+from typing import Any, Callable, Tuple, TYPE_CHECKING
+from urllib.parse import urlparse
 
 import requests
 from configparser import NoOptionError, NoSectionError
@@ -34,7 +35,7 @@ from rucio.common.cache import make_region_memcached
 from rucio.common.config import config_get, config_get_bool
 from rucio.common.constants import FTS_JOB_TYPE, FTS_STATE, FTS_COMPLETE_STATE
 from rucio.common.exception import TransferToolTimeout, TransferToolWrongAnswer, DuplicateFileTransferSubmission
-from rucio.common.utils import APIEncoder, chunks, set_checksum_value
+from rucio.common.utils import APIEncoder, chunks, PREFERRED_CHECKSUM
 from rucio.core.request import get_source_rse, get_transfer_error
 from rucio.core.rse import get_rse_supported_checksums_from_attributes
 from rucio.core.oidc import get_token_for_account_operation
@@ -42,6 +43,9 @@ from rucio.core.monitor import record_counter, record_timer, MultiCounter
 from rucio.transfertool.transfertool import Transfertool, TransferToolBuilder, TransferStatusReport
 from rucio.db.sqla.constants import RequestState
 
+if TYPE_CHECKING:
+    from rucio.core.transfer import DirectTransferDefinition
+
 logging.getLogger("requests").setLevel(logging.CRITICAL)
 disable_warnings()
 
@@ -74,6 +78,27 @@ FINAL_FTS_JOB_STATES = (FTS_STATE.FAILED, FTS_STATE.CANCELED, FTS_STATE.FINISHED
 FINAL_FTS_FILE_STATES = (FTS_STATE.FAILED, FTS_STATE.CANCELED, FTS_STATE.FINISHED, FTS_STATE.NOT_USED)
 
 
+def _configured_source_strategy(activity: str, logger: Callable[..., Any]) -> str:
+    """
+    Retrieve from the configuration the source selection strategy for the given activity
+    """
+    try:
+        default_source_strategy = config_get(section='conveyor', option='default-source-strategy')
+    except (NoOptionError, NoSectionError, RuntimeError):
+        default_source_strategy = 'orderly'
+
+    try:
+        activity_source_strategy = config_get(section='conveyor', option='activity-source-strategy')
+        activity_source_strategy = loads(activity_source_strategy)
+    except (NoOptionError, NoSectionError, RuntimeError):
+        activity_source_strategy = {}
+    except ValueError:
+        logger(logging.WARNING, 'activity_source_strategy not properly defined')
+        activity_source_strategy = {}
+
+    return activity_source_strategy.get(str(activity), default_source_strategy)
+
+
 def oidc_supported(transfer_hop) -> bool:
     """
     checking OIDC AuthN/Z support per destination and source RSEs;
@@ -90,59 +115,52 @@ def oidc_supported(transfer_hop) -> bool:
     return True
 
 
-def checksum_validation_strategy(src_attributes, dst_attributes, logger):
+def _checksum_validation_strategy(
+    transfer: "DirectTransferDefinition",
+    logger: Callable[..., Any],
+) -> Tuple[str, str]:
     """
-    Compute the checksum validation strategy (none, source, destination or both) and the
-    supported checksums from the attributes of the source and destination RSE.
+    Compute the checksum validation strategy (none, source, destination or both) and the checksum
+    to use (in the format accepted by FTS) from the attributes of the source and destination RSE.
     """
-    source_supported_checksums = get_rse_supported_checksums_from_attributes(src_attributes)
-    dest_supported_checksums = get_rse_supported_checksums_from_attributes(dst_attributes)
-    common_checksum_names = set(source_supported_checksums).intersection(dest_supported_checksums)
-
-    verify_checksum = 'both'
-    if not dst_attributes.get('verify_checksum', True):
-        if not src_attributes.get('verify_checksum', True):
-            verify_checksum = 'none'
-        else:
-            verify_checksum = 'source'
+    src_attributes = transfer.src.rse.attributes
+    if src_attributes.get('verify_checksum', True):
+        src_checksums = set(get_rse_supported_checksums_from_attributes(src_attributes))
     else:
-        if not src_attributes.get('verify_checksum', True):
-            verify_checksum = 'destination'
-        else:
-            verify_checksum = 'both'
+        src_checksums = set()
 
-    if len(common_checksum_names) == 0:
-        logger(logging.INFO, 'No common checksum method. Verifying destination only.')
-        verify_checksum = 'destination'
+    dst_attributes = transfer.dst.rse.attributes
+    if dst_attributes.get('verify_checksum', True):
+        dst_checksums = set(get_rse_supported_checksums_from_attributes(dst_attributes))
+    else:
+        dst_checksums = set()
 
-    if source_supported_checksums == ['none']:
-        if dest_supported_checksums == ['none']:
-            # both endpoints support none
-            verify_checksum = 'none'
-        else:
-            # src supports none but dst does
-            verify_checksum = 'destination'
+    intersection = src_checksums.intersection(dst_checksums)
+
+    if intersection:
+        strategy, possible_checksums = 'both', intersection
+    elif dst_checksums:
+        # The prioritization of destination over source here is desired, not random
+        logger(logging.INFO, 'No common checksum method. Verifying destination only.')
+        strategy, possible_checksums = 'target', dst_checksums
+    elif src_checksums:
+        logger(logging.INFO, 'No common checksum method. Verifying source only.')
+        strategy, possible_checksums = 'source', src_checksums
     else:
-        if dest_supported_checksums == ['none']:
-            # source supports some but destination does not
-            verify_checksum = 'source'
-        else:
-            if len(common_checksum_names) == 0:
-                # source and dst support some bot none in common (dst priority)
-                verify_checksum = 'destination'
-            else:
-                # Don't override the value in the file_metadata
-                pass
+        logger(logging.INFO, 'No common checksum method. Not verifying source nor destination.')
+        strategy, possible_checksums = 'none', set()
 
-    checksums_to_use = ['none']
-    if verify_checksum == 'both':
-        checksums_to_use = common_checksum_names
-    elif verify_checksum == 'source':
-        checksums_to_use = source_supported_checksums
-    elif verify_checksum == 'destination':
-        checksums_to_use = dest_supported_checksums
+    checksum_to_use = None
+    for checksum_name in possible_checksums:
+        checksum_value = getattr(transfer.rws, checksum_name, '')
+        if not checksum_value:
+            continue
 
-    return verify_checksum, checksums_to_use
+        checksum_to_use = '%s:%s' % (checksum_name.upper(), checksum_value)
+        if checksum_name == PREFERRED_CHECKSUM:
+            break
+
+    return strategy, checksum_to_use
 
 
 def job_params_for_fts_transfer(transfer, bring_online, default_lifetime, archive_timeout_override, max_time_in_queue, logger, multihop=False):
@@ -167,8 +185,7 @@ def job_params_for_fts_transfer(transfer, bring_online, default_lifetime, archiv
     strict_copy = transfer.dst.rse.attributes.get('strict_copy', False)
     archive_timeout = transfer.dst.rse.attributes.get('archive_timeout', None)
 
-    verify_checksum, checksums_to_use = checksum_validation_strategy(transfer.src.rse.attributes, transfer.dst.rse.attributes, logger=logger)
-    transfer['checksums_to_use'] = checksums_to_use
+    verify_checksum, _checksum_to_use = _checksum_validation_strategy(transfer, logger=logger)
 
     job_params = {'account': transfer.rws.account,
                   'verify_checksum': verify_checksum,
@@ -233,24 +250,6 @@ def bulk_group_transfers(transfer_paths, policy='rule', group_bulk=200, source_s
     grouped_transfers = {}
     grouped_jobs = []
 
-    try:
-        default_source_strategy = config_get(section='conveyor', option='default-source-strategy')
-    except (NoOptionError, NoSectionError, RuntimeError):
-        default_source_strategy = 'orderly'
-
-    try:
-        activity_source_strategy = config_get(section='conveyor', option='activity-source-strategy')
-        activity_source_strategy = loads(activity_source_strategy)
-    except (NoOptionError, NoSectionError, RuntimeError):
-        activity_source_strategy = {}
-    except ValueError:
-        logger(logging.WARNING, 'activity_source_strategy not properly defined')
-        activity_source_strategy = {}
-
-    for transfer_path in transfer_paths:
-        for i, transfer in enumerate(transfer_path):
-            transfer['selection_strategy'] = source_strategy if source_strategy else activity_source_strategy.get(str(transfer.rws.activity), default_source_strategy)
-
     _build_job_params = functools.partial(job_params_for_fts_transfer,
                                           bring_online=bring_online,
                                           default_lifetime=default_lifetime,
@@ -725,9 +724,9 @@ class FTS3Transfertool(Transfertool):
         )
         return jobs
 
-    @classmethod
-    def __file_from_transfer(cls, transfer, job_params):
+    def __file_from_transfer(self, transfer):
         rws = transfer.rws
+        _verify_checksum, checksum_to_use = _checksum_validation_strategy(transfer, logger=self.logger)
         t_file = {
             'sources': [s[1] for s in transfer.legacy_sources],
             'destinations': [transfer.dest_url],
@@ -748,14 +747,10 @@ class FTS3Transfertool(Transfertool):
                 'adler32': rws.adler32
             },
             'filesize': rws.byte_count,
-            'checksum': None,
-            'verify_checksum': job_params['verify_checksum'],
-            'selection_strategy': transfer['selection_strategy'],
-            'request_type': rws.request_type,
+            'checksum': checksum_to_use,
+            'selection_strategy': self.source_strategy if self.source_strategy else _configured_source_strategy(transfer.rws.activity, logger=self.logger),
             'activity': rws.activity
         }
-        if t_file['verify_checksum'] != 'none':
-            set_checksum_value(t_file, transfer['checksums_to_use'])
         return t_file
 
     def submit(self, transfers, job_params, timeout=None):
@@ -776,7 +771,7 @@ class FTS3Transfertool(Transfertool):
                 files.append(transfer)
                 continue
 
-            files.append(self.__file_from_transfer(transfer, job_params))
+            files.append(self.__file_from_transfer(transfer))
 
         # FTS3 expects 'davs' as the scheme identifier instead of https
         for transfer_file in files:
['lib/rucio/transfertool/fts3.py', 'lib/rucio/tests/test_conveyor.py', 'lib/rucio/core/rse.py', 'lib/rucio/core/transfer.py', 'lib/rucio/common/utils.py']