Problem with multihop transfer between 2 cloud RSEs
### Description

Transfers between 2 clouds endpoints is currently not supported and need to be hopped though an intermediate RSE. In this example : https://fts3-pilot.cern.ch:8449/fts3/ftsmon/#/job/b568e444-8db4-11ed-8b35-fa163ec6a1a1 the hop doesn't respect the `verify_checksum: False` for the initial RSE

### Steps to reproduce

...

### Rucio Version

_No response_

### Additional Information

_No response_
diff --git a/lib/rucio/tests/test_conveyor.py b/lib/rucio/tests/test_conveyor.py
index cf7e95cf9..6b13fc679 100644
--- a/lib/rucio/tests/test_conveyor.py
+++ b/lib/rucio/tests/test_conveyor.py
@@ -1151,8 +1151,8 @@ class FTSWrapper(FTS3Transfertool):
     def on_receive(job_response):
         pass
 
-    def _FTS3Transfertool__file_from_transfer(self, transfer):
-        file = super()._FTS3Transfertool__file_from_transfer(transfer)
+    def _file_from_transfer(self, transfer, job_params):
+        file = super()._file_from_transfer(transfer, job_params)
         self.on_submit(file)
         return file
 
diff --git a/lib/rucio/transfertool/fts3.py b/lib/rucio/transfertool/fts3.py
index 4ef6e86bd..af159c025 100644
--- a/lib/rucio/transfertool/fts3.py
+++ b/lib/rucio/transfertool/fts3.py
@@ -15,7 +15,6 @@
 
 import datetime
 import json
-import functools
 import logging
 import traceback
 import uuid
@@ -44,6 +43,8 @@ from rucio.transfertool.transfertool import Transfertool, TransferToolBuilder, T
 from rucio.db.sqla.constants import RequestState
 
 if TYPE_CHECKING:
+    from typing import List, Set, Union
+
     from rucio.core.transfer import DirectTransferDefinition
     from rucio.core.rse import RseData
 
@@ -79,6 +80,30 @@ REWRITE_HTTPS_TO_DAVS = config_get_bool('transfers', 'rewrite_https_to_davs', de
 FINAL_FTS_JOB_STATES = (FTS_STATE.FAILED, FTS_STATE.CANCELED, FTS_STATE.FINISHED, FTS_STATE.FINISHEDDIRTY)
 FINAL_FTS_FILE_STATES = (FTS_STATE.FAILED, FTS_STATE.CANCELED, FTS_STATE.FINISHED, FTS_STATE.NOT_USED)
 
+# In a multi-hop transfer, we must compute a checksum validation strategy valid for the whole path.
+# This state-machine defines how strategies of hops are merged into a path-wide strategy.
+# For example, if HOP1 supports only validation of checksum at source while HOP2 only
+# supports validation at destination, the strategy for the whole path MUST be "none". Otherwise,
+# transfers will fail when FTS will try to validate the checksum.
+PATH_CHECKSUM_VALIDATION_STRATEGY: "Dict[Tuple[str, str], str]" = {
+    ('both', 'both'): 'both',
+    ('both', 'target'): 'target',
+    ('both', 'source'): 'source',
+    ('both', 'none'): 'none',
+    ('target', 'both'): 'target',
+    ('target', 'target'): 'target',
+    ('target', 'source'): 'none',
+    ('target', 'none'): 'none',
+    ('source', 'both'): 'source',
+    ('source', 'target'): 'none',
+    ('source', 'source'): 'source',
+    ('source', 'none'): 'none',
+    ('none', 'both'): 'none',
+    ('none', 'target'): 'none',
+    ('none', 'source'): 'none',
+    ('none', 'none'): 'none',
+}
+
 
 def _configured_source_strategy(activity: str, logger: Callable[..., Any]) -> str:
     """
@@ -117,13 +142,11 @@ def oidc_supported(transfer_hop) -> bool:
     return True
 
 
-def _checksum_validation_strategy(
-    transfer: "DirectTransferDefinition",
-    logger: Callable[..., Any],
-) -> Tuple[str, str]:
+def _available_checksums(
+        transfer: "DirectTransferDefinition",
+) -> "Tuple[Set[str], Set[str]]":
     """
-    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.
+    Get checksums which can be used for file validation on the source and the destination RSE
     """
     src_attributes = transfer.src.rse.attributes
     if src_attributes.get('verify_checksum', True):
@@ -137,20 +160,73 @@ def _checksum_validation_strategy(
     else:
         dst_checksums = set()
 
+    return src_checksums, dst_checksums
+
+
+def _hop_checksum_validation_strategy(
+        transfer: "DirectTransferDefinition",
+        logger: Callable[..., Any],
+) -> "Tuple[str, Set[str]]":
+    """
+    Compute the checksum validation strategy (none, source, destination or both) depending
+    on available source and destination checksums for a single hop transfer
+    """
+    src_checksums, dst_checksums = _available_checksums(transfer)
     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.')
+        logger(logging.INFO, f'No common checksum method for {transfer}. Verifying destination only.')
         strategy, possible_checksums = 'target', dst_checksums
     elif src_checksums:
-        logger(logging.INFO, 'No common checksum method. Verifying source only.')
+        logger(logging.INFO, f'No common checksum method for {transfer}. Verifying source only.')
         strategy, possible_checksums = 'source', src_checksums
     else:
-        logger(logging.INFO, 'No common checksum method. Not verifying source nor destination.')
+        logger(logging.INFO, f'No common checksum method for {transfer}. Not verifying source nor destination.')
         strategy, possible_checksums = 'none', set()
+    return strategy, possible_checksums
+
+
+def _path_checksum_validation_strategy(
+        transfer_path: "List[DirectTransferDefinition]",
+        logger: Callable[..., Any],
+) -> str:
+    """
+    Compute the checksum validation strategy for the whole transfer path.
+    """
+
+    path_strategy = 'both'
+    for transfer_hop in transfer_path:
+        hop_strategy, _ = _hop_checksum_validation_strategy(transfer_hop, logger)
+
+        path_strategy = PATH_CHECKSUM_VALIDATION_STRATEGY.get((path_strategy, hop_strategy), 'none')
+
+    return path_strategy
+
+
+def _pick_fts_checksum(
+        transfer: "DirectTransferDefinition",
+        path_strategy: "str",
+) -> "Union[str, None]":
+    """
+    Pick the checksum to use for validating file integrity on this particular transfer hop.
+    This function will only work correctly for values of 'path_strategy' which are
+    valid for the englobing multi-hop transfer path.
+
+    Returns the checksum as a string in the format expected by the FTS bulks submission API.
+    """
+    src_checksums, dst_checksums = _available_checksums(transfer)
+
+    if path_strategy == 'both':
+        possible_checksums = src_checksums.intersection(dst_checksums)
+    elif path_strategy == 'target':
+        possible_checksums = dst_checksums
+    elif path_strategy == 'source':
+        possible_checksums = src_checksums
+    else:
+        possible_checksums = set()
 
     checksum_to_use = None
     for checksum_name in possible_checksums:
@@ -162,69 +238,71 @@ def _checksum_validation_strategy(
         if checksum_name == PREFERRED_CHECKSUM:
             break
 
-    return strategy, checksum_to_use
+    return checksum_to_use
 
 
-def job_params_for_fts_transfer(transfer, bring_online, default_lifetime, archive_timeout_override, max_time_in_queue, logger, multihop=False):
+def build_job_params(transfer_path, bring_online, default_lifetime, archive_timeout_override, max_time_in_queue, logger):
     """
     Prepare the job parameters which will be passed to FTS transfertool
     """
 
+    # The last hop is the main request (the one which triggered the whole transfer),
+    # so most attributes will come from it
+    last_hop = transfer_path[-1]
+    first_hop = transfer_path[0]
+
     overwrite, bring_online_local = True, None
-    if transfer.src.rse.is_tape_or_staging_required():
+    if first_hop.src.rse.is_tape_or_staging_required():
+        # Activate bring_online if it was requested by first hop
+        # We don't allow multihop via a tape, so bring_online should not be set on any other hop
         bring_online_local = bring_online
-    if transfer.dst.rse.is_tape():
+    if last_hop.dst.rse.is_tape():
         overwrite = False
 
     # Get dest space token
-    dest_protocol = transfer.protocol_factory.protocol(transfer.dst.rse, transfer.dst.scheme, transfer.operation_dest)
+    dest_protocol = last_hop.protocol_factory.protocol(last_hop.dst.rse, last_hop.dst.scheme, last_hop.operation_dest)
     dest_spacetoken = None
     if dest_protocol.attributes and 'extended_attributes' in dest_protocol.attributes and \\
             dest_protocol.attributes['extended_attributes'] and 'space_token' in dest_protocol.attributes['extended_attributes']:
         dest_spacetoken = dest_protocol.attributes['extended_attributes']['space_token']
-    src_spacetoken = None
 
-    strict_copy = transfer.dst.rse.attributes.get('strict_copy', False)
-    archive_timeout = transfer.dst.rse.attributes.get('archive_timeout', None)
-    src_rse_s3_url_style = transfer.src.rse.attributes.get('s3_url_style', None)
-    dst_rse_s3_url_style = transfer.dst.rse.attributes.get('s3_url_style', None)
+    strict_copy = last_hop.dst.rse.attributes.get('strict_copy', False)
+    archive_timeout = last_hop.dst.rse.attributes.get('archive_timeout', None)
 
-    verify_checksum, _checksum_to_use = _checksum_validation_strategy(transfer, logger=logger)
-
-    job_params = {'account': transfer.rws.account,
-                  'verify_checksum': verify_checksum,
-                  'copy_pin_lifetime': transfer.rws.attributes.get('lifetime', default_lifetime),
+    job_params = {'account': last_hop.rws.account,
+                  'verify_checksum': _path_checksum_validation_strategy(transfer_path, logger=logger),
+                  'copy_pin_lifetime': last_hop.rws.attributes.get('lifetime', default_lifetime),
                   'bring_online': bring_online_local,
                   'job_metadata': {
                       'issuer': 'rucio',
-                      'multi_sources': True if len(transfer.legacy_sources) > 1 else False,
+                      'multi_sources': False,
                   },
-                  'overwrite': transfer.rws.attributes.get('overwrite', overwrite),
-                  'priority': transfer.rws.priority}
+                  'overwrite': last_hop.rws.attributes.get('overwrite', overwrite),
+                  'priority': last_hop.rws.priority}
 
-    if multihop:
+    if len(transfer_path) > 1:
         job_params['multihop'] = True
         job_params['job_metadata']['multihop'] = True
+    elif len(last_hop.legacy_sources) > 1:
+        job_params['job_metadata']['multi_sources'] = True
     if strict_copy:
         job_params['strict_copy'] = strict_copy
     if dest_spacetoken:
         job_params['spacetoken'] = dest_spacetoken
-    if src_spacetoken:
-        job_params['source_spacetoken'] = src_spacetoken
-    if transfer.use_ipv4:
+    if last_hop.use_ipv4:
         job_params['ipv4'] = True
         job_params['ipv6'] = False
 
     # assume s3alternate True (path-style URL S3 RSEs)
     job_params['s3alternate'] = True
-    if src_rse_s3_url_style:
-        if src_rse_s3_url_style == "host":
-            job_params['s3alternate'] = False
-    if dst_rse_s3_url_style:
-        if dst_rse_s3_url_style == "host":
-            job_params['s3alternate'] = False
-
-    if archive_timeout and transfer.dst.rse.is_tape():
+    src_rse_s3_url_style = first_hop.src.rse.attributes.get('s3_url_style', None)
+    if src_rse_s3_url_style == "host":
+        job_params['s3alternate'] = False
+    dst_rse_s3_url_style = last_hop.dst.rse.attributes.get('s3_url_style', None)
+    if dst_rse_s3_url_style == "host":
+        job_params['s3alternate'] = False
+
+    if archive_timeout and last_hop.dst.rse.is_tape():
         try:
             archive_timeout = int(archive_timeout)
             if archive_timeout_override is None:
@@ -235,13 +313,24 @@ def job_params_for_fts_transfer(transfer, bring_online, default_lifetime, archiv
             job_params['dst_file_report'] = True
             logger(logging.DEBUG, 'Added archive timeout to transfer.')
         except ValueError:
-            logger(logging.WARNING, 'Could not set archive_timeout for %s. Must be integer.', transfer)
+            logger(logging.WARNING, 'Could not set archive_timeout for %s. Must be integer.', last_hop)
             pass
     if max_time_in_queue:
-        if transfer.rws.activity in max_time_in_queue:
-            job_params['max_time_in_queue'] = max_time_in_queue[transfer.rws.activity]
+        if last_hop.rws.activity in max_time_in_queue:
+            job_params['max_time_in_queue'] = max_time_in_queue[last_hop.rws.activity]
         elif 'default' in max_time_in_queue:
             job_params['max_time_in_queue'] = max_time_in_queue['default']
+
+    overwrite_hop = True
+    for transfer_hop in transfer_path[:-1]:
+        # Only allow overwrite if all hops in multihop allow it
+        h_overwrite = transfer_hop.rws.attributes.get('overwrite', True)
+        job_params['overwrite'] = h_overwrite and job_params['overwrite']
+        # Allow overwrite_hop if all intermediate hops allow it (ignoring the last hop)
+        overwrite_hop = h_overwrite and overwrite_hop
+    if not job_params['overwrite'] and overwrite_hop:
+        job_params['overwrite_hop'] = overwrite_hop
+
     return job_params
 
 
@@ -261,52 +350,35 @@ def bulk_group_transfers(transfer_paths, policy='rule', group_bulk=200, source_s
     """
 
     grouped_transfers = {}
-    grouped_jobs = []
-
-    _build_job_params = functools.partial(job_params_for_fts_transfer,
-                                          bring_online=bring_online,
-                                          default_lifetime=default_lifetime,
-                                          archive_timeout_override=archive_timeout_override,
-                                          max_time_in_queue=max_time_in_queue,
-                                          logger=logger)
+    fts_jobs = []
+
     for transfer_path in transfer_paths:
-        if len(transfer_path) > 1:
-            # for multihop transfers, all the path is submitted as a separate job
-            job_params = _build_job_params(transfer_path[-1], multihop=True)
-
-            overwrite_hop = True
-            for transfer in transfer_path[:-1]:
-                hop_params = _build_job_params(transfer, multihop=True)
-                # Only allow overwrite if all transfers in multihop allow it
-                job_params['overwrite'] = hop_params['overwrite'] and job_params['overwrite']
-                # Allow overwrite_hop if all intermediate hops allow it (ignoring the last hop)
-                overwrite_hop = hop_params['overwrite'] and overwrite_hop
-                # Activate bring_online if it was requested by first hop (it is a multihop starting at a tape)
-                # We don't allow multihop via a tape, so bring_online should not be set on any other hop
-                if transfer is transfer_path[0] and hop_params['bring_online']:
-                    job_params['bring_online'] = hop_params['bring_online']
-            if not job_params['overwrite'] and overwrite_hop:
-                job_params['overwrite_hop'] = overwrite_hop
-
-            group_key = 'multihop_%s' % transfer_path[-1].rws.request_id
-            grouped_transfers[group_key] = {'transfers': transfer_path[0:group_bulk], 'job_params': job_params}
-        elif len(transfer_path[0].legacy_sources) > 1:
-            # for multi-source transfers, no bulk submission.
-            transfer = transfer_path[0]
-            grouped_jobs.append({'transfers': [transfer], 'job_params': _build_job_params(transfer)})
+        job_params = build_job_params(
+            transfer_path=transfer_path,
+            bring_online=bring_online,
+            default_lifetime=default_lifetime,
+            archive_timeout_override=archive_timeout_override,
+            max_time_in_queue=max_time_in_queue,
+            logger=logger
+        )
+        if job_params['job_metadata'].get('multi_sources') or job_params['job_metadata'].get('multihop'):
+            # for multi-hop and multi-source transfers, no bulk submission.
+            fts_jobs.append({'transfers': transfer_path[0:group_bulk], 'job_params': job_params})
         else:
             # it's a single-hop, single-source, transfer. Hence, a candidate for bulk submission.
             transfer = transfer_path[0]
-            job_params = _build_job_params(transfer)
 
             # we cannot group transfers together if their job_key differ
-            job_key = '%s,%s,%s,%s,%s,%s,%s,%s' % (job_params['verify_checksum'], job_params.get('spacetoken', None),
-                                                   job_params['copy_pin_lifetime'],
-                                                   job_params['bring_online'], job_params['job_metadata'],
-                                                   job_params.get('source_spacetoken', None),
-                                                   job_params['overwrite'], job_params['priority'])
-            if 'max_time_in_queue' in job_params:
-                job_key = job_key + ',%s' % job_params['max_time_in_queue']
+            job_key = '%s,%s,%s,%s,%s,%s,%s,%s' % (
+                job_params['verify_checksum'],
+                job_params.get('spacetoken', ''),
+                job_params['copy_pin_lifetime'],
+                job_params['bring_online'],
+                job_params['job_metadata'],
+                job_params['overwrite'],
+                job_params['priority'],
+                job_params.get('max_time_in_queue', '')
+            )
 
             # Additionally, we don't want to group transfers together if their policy_key differ
             policy_key = ''
@@ -335,9 +407,9 @@ def bulk_group_transfers(transfer_paths, policy='rule', group_bulk=200, source_s
     for group in grouped_transfers.values():
         job_params = group['job_params']
         for transfer_paths in chunks(group['transfers'], group_bulk):
-            grouped_jobs.append({'transfers': transfer_paths, 'job_params': job_params})
+            fts_jobs.append({'transfers': transfer_paths, 'job_params': job_params})
 
-    return grouped_jobs
+    return fts_jobs
 
 
 class Fts3TransferStatusReport(TransferStatusReport):
@@ -769,9 +841,9 @@ class FTS3Transfertool(Transfertool):
         )
         return jobs
 
-    def __file_from_transfer(self, transfer):
+    def _file_from_transfer(self, transfer, job_params):
         rws = transfer.rws
-        _verify_checksum, checksum_to_use = _checksum_validation_strategy(transfer, logger=self.logger)
+        checksum_to_use = _pick_fts_checksum(transfer, path_strategy=job_params['verify_checksum'])
         t_file = {
             'sources': [s[1] for s in transfer.legacy_sources],
             'destinations': [transfer.dest_url],
@@ -815,7 +887,7 @@ class FTS3Transfertool(Transfertool):
                 files.append(transfer)
                 continue
 
-            files.append(self.__file_from_transfer(transfer))
+            files.append(self._file_from_transfer(transfer, job_params))
 
         # 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']