[FLINK-39969][runtime] use the same SSL protcol and algorithms in flink

[JozsefKuti]

What is the purpose of the change

Use the same SSL protcol and algorithms configuration properties in all the flink services and all clients with additional debug logs to make ssl setup visible if needed.

In our flink environment we have to support custom security.ssl.protocol and security.ssl.algorithms configuration, so we had to test through from Tlsv1.2 to Tlsv1.2,Tlsv1.3 till pure Tlsv1.3 tls protocol setups using default and non-default ciphers sets.

In the cases where there was a common set between the configured and the default java ssl.protocol-ssl.algorithms setup, then everything worked fine, there was a tls version and cipher to use between the client and service side.
However when the cases where there was not any common set we faced

• BlobClient cannot connect to its service:
  • org.apache.flink.runtime.blob.Connection - Error wBlobServerhile executing BLOB connection from /0:0:0:0:0:0:0:1:41630.
• MiniDispatcherRestEndpoint fails with: javax.net.ssl.SSLHandshakeException: no cipher suites in common

Brief change log

org.apache.flink.runtime.net.SSLUtils needs the following adjustments:

• SSLUtils#createSSLClientSocketFactory does not sets the enabled protocols and cipherSuites

  • while its service side SSLUtils#createSSLServerSocketFactory does
  • resulting BlobServerConnection-s were dropping its clients
    • since server expects the configured non-default cipher, client was using the default java ciphers

• SSLUtils#createRestNettySSLContext does sets the enabled sslProtocols and ciphers ONLY for the clients and skips it for the services

  • Resulting Flink dashboards - MiniDispatcherRestEndpoint - uses incompatible ciphers
    • one uses the java default tls config - service,
    • one uses the non-defaults the configured one - client

Verifying this change

org.apache.flink.runtime.net.SSLUtilsTest had been extended with the following test cases - using TDD where the new tests were failing without the code fixes:

• testRESTServerSSLConfigCipherAlgorithmsWithoutMutualAuth
• testRESTServerSSLEngineUsesConfiguredProtocolsAndCipherSuites
• testSetSSLVersionAndCipherSuitesForSSLClientSocket

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no

[flinkbot]

CI report:

• bfb8062 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[spuru9]

Hi
Can you add the AI declaration to the PR Description.

[spuru9]

Can you tak a look at some minor nits.

[spuru9]

Suggested change

	private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(SSLUtils.class);
	private static final Logger LOG = LoggerFactory.getLogger(SSLUtils.class);

Keep import alongside others.

[spuru9]

These four near-identical blocks are a lot of noise for diagnostic logging. Consider a tiny private helper like logSslDiag(String scope, String[] protocols, Iterable ciphers).
Also the [SSL-DIAG] bracket-prefix style isn't used elsewhere in Flink loggin

[spuru9]

nit: drop the fully-qualified names (javax.net.ssl.SSLSocket, java.net.Socket, etc.) in favor of import