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

Code before:
                            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<
                        RestfulGateway, EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters>
                testRestHandler =
                        new TestRestHandler<>(
                                gatewayRetriever,
                                TEST_MESSAGE_HEADERS,
                                CompletableFuture.completedFuture(EmptyResponseBody.getInstance()));

        return TestRestServerEndpoint.builder(config)
                .withHandler(TEST_MESSAGE_HEADERS, testRestHandler)
                .buildAndStart();
    }

    private RestClient getRestClient(String host, int port) throws ConfigurationException {
        return new RestClient(config, Executors.directExecutor(), host, port, null);
    }
}

Review comment (bug):
Can we add something like `message.contains("hostname verification failed")` because cert verification can blow up many ways?

Code after:
        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);

            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))
                    .withMessageContaining("No subject alternative names matching IP address");
        }
    }

    /** Asserts that the SANs in the certificate are what we expect. */