GH-8674: Fix ServerKeyVerifier impl for key type (#8675)

Fixes #8674

The `ResourceKnownHostsServerKeyVerifier` does not take into account that several different
keys can be present in the known hosts resource for the same host/port

* Fix `ResourceKnownHostsServerKeyVerifier` to find a list of knows host for the requested session.
Then iterate of this result to match the key type first and then compare keys and their `revoked` marker

**Cherry-pick to `6.1.x` & `6.0.x`**

Author: artembilan

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/ResourceKnownHostsServerKeyVerifier.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -66,18 +66,19 @@ public ResourceKnownHostsServerKeyVerifier(Resource knownHostsResource) {
 	@Override
 	public boolean verifyServerKey(ClientSession clientSession, SocketAddress remoteAddress, PublicKey serverKey) {
 		Collection<KnownHostsServerKeyVerifier.HostEntryPair> knownHosts = this.keysSupplier.get();
-		KnownHostsServerKeyVerifier.HostEntryPair match = findKnownHostEntry(clientSession, remoteAddress, knownHosts);
-		if (match == null) {
+		List<KnownHostsServerKeyVerifier.HostEntryPair> matches =
+				findKnownHostEntries(clientSession, remoteAddress, knownHosts);
+
+		if (matches.isEmpty()) {
 			return false;
 		}
 
-		KnownHostEntry entry = match.getHostEntry();
-		PublicKey expected = match.getServerKey();
-		if (KeyUtils.compareKeys(expected, serverKey)) {
-			return !"revoked".equals(entry.getMarker());
-		}
+		String serverKeyType = KeyUtils.getKeyType(serverKey);
 
-		return false;
+		return matches.stream()
+				.filter(match -> serverKeyType.equals(match.getHostEntry().getKeyEntry().getKeyType()))
+				.filter(match -> KeyUtils.compareKeys(match.getServerKey(), serverKey))
+				.anyMatch(match -> !"revoked".equals(match.getHostEntry().getMarker()));
 	}
 
 	private static Supplier<Collection<KnownHostsServerKeyVerifier.HostEntryPair>> getKnownHostSupplier(
@@ -106,26 +107,32 @@ private static PublicKey resolveHostKey(KnownHostEntry entry) throws IOException
 		return authEntry.resolvePublicKey(null, PublicKeyEntryResolver.IGNORING);
 	}
 
-	private static KnownHostsServerKeyVerifier.HostEntryPair findKnownHostEntry(
+	private static List<KnownHostsServerKeyVerifier.HostEntryPair> findKnownHostEntries(
 			ClientSession clientSession, SocketAddress remoteAddress,
 			Collection<KnownHostsServerKeyVerifier.HostEntryPair> knownHosts) {
 
+		if (GenericUtils.isEmpty(knownHosts)) {
+			return Collections.emptyList();
+		}
+
 		Collection<SshdSocketAddress> candidates = resolveHostNetworkIdentities(clientSession, remoteAddress);
 
 		if (GenericUtils.isEmpty(candidates)) {
-			return null;
+			return Collections.emptyList();
 		}
 
+		List<KnownHostsServerKeyVerifier.HostEntryPair> matches = new ArrayList<>();
 		for (KnownHostsServerKeyVerifier.HostEntryPair match : knownHosts) {
 			KnownHostEntry entry = match.getHostEntry();
 			for (SshdSocketAddress host : candidates) {
 				if (entry.isHostMatch(host.getHostName(), host.getPort())) {
-					return match;
+					matches.add(match);
+					break;
 				}
 			}
 		}
 
-		return null; // no match found
+		return matches;
 	}
 
 	private static Collection<SshdSocketAddress> resolveHostNetworkIdentities(

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java

@@ -19,6 +19,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.net.ConnectException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -121,7 +122,7 @@ public void concurrentGetSessionDoesntCauseFailure() throws IOException {
 				asyncTaskExecutor.execute(() -> concurrentSessions.add(sftpSessionFactory.getSession()));
 			}
 
-			await().until(() -> concurrentSessions.size() == 3);
+			await().atMost(Duration.ofSeconds(30)).until(() -> concurrentSessions.size() == 3);
 
 			assertThat(concurrentSessions.get(0))
 					.isNotEqualTo(concurrentSessions.get(1))