Repository: apache/flink
File: flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestSSLEndpointVerificationITCase.java (Java)

Code before:
        config.set(SecurityOptions.SSL_REST_KEYSTORE, KEY_STORE_FILE.getAbsolutePath());
        config.set(SecurityOptions.SSL_REST_KEYSTORE_PASSWORD, "password");
        config.set(SecurityOptions.SSL_REST_KEY_PASSWORD, "password");
        config.set(SecurityOptions.SSL_REST_VERIFY_HOSTNAME, true);
    }

    /**
     * With hostname verification turned on, this should execute fine as "127.0.0.1" is part of the
     * Subject Alternative Names.
     */
    @Test
    void testConnectSuccess() throws Exception {
        try (RestServerEndpoint serverEndpoint = getRestServerEndpoint();
                RestClient restClient = getRestClient("127.0.0.1", serverEndpoint.getRestPort())) {
            restClient
                    .sendRequest(
                            serverEndpoint.getServerAddress().getHostName(),
                            serverEndpoint.getServerAddress().getPort(),
                            TEST_MESSAGE_HEADERS,
                            EmptyMessageParameters.getInstance(),
                            EmptyRequestBody.getInstance(),
                            Collections.emptyList())
                    .get();
        }
    }

    /** This should fail as "127.0.0.2" is not part of the Subject Alternative Names. */
    @Test
    void testConnectFailure() throws Exception {
        try (RestServerEndpoint serverEndpoint = getRestServerEndpoint();
                RestClient restClient = getRestClient("127.0.0.2", serverEndpoint.getRestPort())) {
            assertThatFuture(
                            restClient.sendRequest(
                                    serverEndpoint.getServerAddress().getHostName(),
                                    serverEndpoint.getServerAddress().getPort(),
                                    TEST_MESSAGE_HEADERS,
                                    EmptyMessageParameters.getInstance(),
                                    EmptyRequestBody.getInstance(),
                                    Collections.emptyList()))
                    .failsWithin(60, TimeUnit.SECONDS)
                    .withThrowableOfType(ExecutionException.class)
                    .satisfies(anyCauseMatches(CertificateException.class));
        }
    }

    private RestServerEndpoint getRestServerEndpoint() throws Exception {
        RestfulGateway restfulGateway = new TestingRestfulGateway.Builder().build();
        final GatewayRetriever<RestfulGateway> gatewayRetriever =
                () -> CompletableFuture.completedFuture(restfulGateway);

        final TestRestHandler<

Review comment (refactor):
I always suggest to have at least one assertion on tests. Maybe we can add here something like:
```
          // Extract and verify the certificate SANs
          SSLContext sslContext = SSLContext.getDefault();
          SSLSocketFactory factory = sslContext.getSocketFactory();
          try (SSLSocket socket = (SSLSocket) factory.createSocket(
                  serverEndpoint.getServerAddress().getHostName(),
                  serverEndpoint.getServerAddress().getPort())) {

              socket.startHandshake();
              Certificate[] certs = socket.getSession().getPeerCertificates();
              X509Certificate x509 = (X509Certificate) certs[0];

              // Extract Subject Alternative Names
              Collection<List<?>> sans = x509.getSubjectAlternativeNames();
              List<String> ipAddresses = sans.stream()
                      .filter(san -> (Integer) san.get(0) == 7) // Type 7 is IP address
                      .map(san -> (String) san.get(1))
                      .collect(Collectors.toList());

              // Assert that 127.0.0.1 is in the certificate
              assertThat(ipAddresses).contains("127.0.0.1");
          }
```


Code after:
            TEST_MESSAGE_HEADERS =
                    TestMessageHeaders.emptyBuilder()
                            .setTargetRestEndpointURL("/test-handler")
                            .build();

    private final Configuration config;

    public RestSSLEndpointVerificationITCase() {
        config = new Configuration();
        config.set(RestOptions.BIND_PORT, "0");
        config.set(RestOptions.ADDRESS, "localhost");
        config.set(SecurityOptions.SSL_REST_ENABLED, true);
        config.set(SecurityOptions.SSL_REST_TRUSTSTORE, TRUST_STORE_FILE.getAbsolutePath());
        config.set(SecurityOptions.SSL_REST_TRUSTSTORE_PASSWORD, "password");
        config.set(SecurityOptions.SSL_REST_KEYSTORE, KEY_STORE_FILE.getAbsolutePath());
        config.set(SecurityOptions.SSL_REST_KEYSTORE_PASSWORD, "password");
        config.set(SecurityOptions.SSL_REST_KEY_PASSWORD, "password");
        config.set(SecurityOptions.SSL_REST_VERIFY_HOSTNAME, true);
    }

    /**
     * With hostname verification turned on, this should execute fine as "127.0.0.1" is part of the
     * Subject Alternative Names.
     */
    @Test
    void testConnectSuccess() throws Exception {
        try (RestServerEndpoint serverEndpoint = getRestServerEndpoint();
                RestClient restClient = getRestClient("127.0.0.1", serverEndpoint.getRestPort())) {
            assertSan(serverEndpoint);

            var result =
                    restClient
                            .sendRequest(
                                    serverEndpoint.getServerAddress().getHostName(),
                                    serverEndpoint.getServerAddress().getPort(),
                                    TEST_MESSAGE_HEADERS,
                                    EmptyMessageParameters.getInstance(),
                                    EmptyRequestBody.getInstance(),
                                    Collections.emptyList())
                            .get();
            assertThat(result).isNotNull();
        }
    }

    /** This should fail as "127.0.0.2" is not part of the Subject Alternative Names. */
    @Test
    void testConnectFailure() throws Exception {
        try (RestServerEndpoint serverEndpoint = getRestServerEndpoint();
                RestClient restClient = getRestClient("127.0.0.2", serverEndpoint.getRestPort())) {
            assertSan(serverEndpoint);