/*
 * Copyright 2018, EnMasse authors.
 * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
 */
package io.enmasse.systemtest.common.certs;

import static io.enmasse.systemtest.TestTag.isolated;
import static org.junit.jupiter.api.Assertions.fail;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.security.KeyStore;
import java.security.SecureRandom;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.util.Arrays;
import java.util.Base64;
import java.util.Collections;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;

import io.enmasse.address.model.*;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
import org.slf4j.Logger;

import io.enmasse.systemtest.AddressSpaceType;
import io.enmasse.systemtest.CertBundle;
import io.enmasse.systemtest.CertProvider;
import io.enmasse.systemtest.CustomLogger;
import io.enmasse.systemtest.DestinationPlan;
import io.enmasse.systemtest.Endpoint;
import io.enmasse.systemtest.Environment;
import io.enmasse.systemtest.SystemtestsKubernetesApps;
import io.enmasse.systemtest.TimeoutBudget;
import io.enmasse.systemtest.UserCredentials;
import io.enmasse.systemtest.VertxFactory;
import io.enmasse.systemtest.amqp.AmqpClient;
import io.enmasse.systemtest.apiclients.OpenshiftCertValidatorApiClient;
import io.enmasse.systemtest.bases.TestBase;
import io.enmasse.systemtest.standard.QueueTest;
import io.enmasse.systemtest.utils.AddressSpaceUtils;
import io.enmasse.systemtest.utils.AddressUtils;
import io.enmasse.systemtest.utils.CertificateUtils;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.PemTrustOptions;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;

@Tag(isolated)
class CertProviderTest extends TestBase {

    private static Logger log = CustomLogger.getLogger();
    private static String ENDPOINT_PREFIX = "test-endpoint-";

    private AddressSpace addressSpace;
    private UserCredentials user;
    private Address queue;
    private Address topic;

    @Test
    void testSelfSigned() throws Exception {

        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.selfsigned.name())
                .build());

        String caCert = new String(Base64.getDecoder().decode(getAddressSpace(addressSpace.getMetadata().getName()).getStatus().getCaCert()));

        AmqpClient amqpClient = amqpClientFactory.createQueueClient(addressSpace);
        amqpClient.getConnectOptions().setCredentials(user).setCert(caCert);

        MqttConnectOptions mqttOptions = new MqttConnectOptions();
        mqttOptions.setSocketFactory(getSocketFactory(new ByteArrayInputStream(caCert.getBytes())));
        mqttOptions.setUserName(user.getUsername());
        mqttOptions.setPassword(user.getPassword().toCharArray());
        IMqttClient mqttClient = mqttClientFactory.build()
                .addressSpace(addressSpace)
                .mqttConnectionOptions(mqttOptions).create();

        testCertProvider(amqpClient, mqttClient);
    }

    @Test
    @Disabled("Disabled due to #2427")
    void testConsoleSelfSignedTrustEndpointCert() throws Exception {
        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.selfsigned.name())
                .build(), false);

        String caCert = new String(Base64.getDecoder().decode(getAddressSpace(addressSpace.getMetadata().getName()).getStatus().getCaCert()));

        WebClient webClient = WebClient.create(VertxFactory.create(), new WebClientOptions()
                .setSsl(true)
                .setTrustAll(false)
                .setPemTrustOptions(new PemTrustOptions()
                          .addCertValue(Buffer.buffer(caCert)))
                .setVerifyHost(false));

        testConsole(webClient);
    }

    @Test
    @Disabled("Disabled due to #2427")
    void testConsoleSelfSignedTrustCaCert() throws Exception {
        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.selfsigned.name())
                .build(), false);

        String endpointCert = new String(Base64.getDecoder().decode(
                getAddressSpace(addressSpace.getMetadata().getName())
                        .getSpec()
                        .getEndpoints()
                        .stream()
                        .filter(e->e.getService().equals("console"))
                        .findFirst()
                        .get()
                        .getCert()
                        .getTlsCert()));

        WebClient webClient = WebClient.create(VertxFactory.create(), new WebClientOptions()
                .setSsl(true)
                .setTrustAll(false)
                .setPemTrustOptions(new PemTrustOptions()
                          .addCertValue(Buffer.buffer(endpointCert)))
                .setVerifyHost(false));

        testConsole(webClient);
    }

    @Test
    void testCertBundle() throws Exception {
        CertBundle certBundle = CertificateUtils.createCertBundle();

        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.certBundle.name())
                .withTlsKey(certBundle.getKeyB64())
                .withTlsCert(certBundle.getCertB64())
                .build());

        AmqpClient amqpClient = amqpClientFactory.createQueueClient(addressSpace);
        amqpClient.getConnectOptions()
            .setCredentials(user)
            .getProtonClientOptions()
                .setSsl(true)
                .setHostnameVerificationAlgorithm("")
                .setPemTrustOptions(new PemTrustOptions()
                        .addCertValue(Buffer.buffer(certBundle.getCaCert())))
                .setTrustAll(false);

        MqttConnectOptions mqttOptions = new MqttConnectOptions();
        mqttOptions.setSocketFactory(getSocketFactory(new ByteArrayInputStream(certBundle.getCaCert().getBytes())));
        mqttOptions.setUserName(user.getUsername());
        mqttOptions.setPassword(user.getPassword().toCharArray());
        IMqttClient mqttClient = mqttClientFactory.build()
                .addressSpace(addressSpace)
                .mqttConnectionOptions(mqttOptions).create();

        testCertProvider(amqpClient, mqttClient);

    }

    @Test
    @Disabled("Disabled due to #2427")
    void testConsoleCertBundleTrustEndpointCert() throws Exception {
        CertBundle certBundle = CertificateUtils.createCertBundle();

        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.certBundle.name())
                .withTlsKey(certBundle.getKeyB64())
                .withTlsCert(certBundle.getCertB64())
                .build(), false);

        WebClient webClient = WebClient.create(VertxFactory.create(), new WebClientOptions()
                .setSsl(true)
                .setTrustAll(false)
                .setPemTrustOptions(new PemTrustOptions()
                        .addCertValue(Buffer.buffer(certBundle.getCert())))
                .setVerifyHost(false));

        testConsole(webClient);
    }

    @Test
    @Disabled("Disabled due to #2427")
    void testConsoleCertBundleTrustCaCert() throws Exception {
        CertBundle certBundle = CertificateUtils.createCertBundle();

        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.certBundle.name())
                .withTlsKey(certBundle.getKeyB64())
                .withTlsCert(certBundle.getCertB64())
                .build(), false);

        WebClient webClient = WebClient.create(VertxFactory.create(), new WebClientOptions()
                .setSsl(true)
                .setTrustAll(false)
                .setPemTrustOptions(new PemTrustOptions()
                        .addCertValue(Buffer.buffer(certBundle.getCaCert())))
                .setVerifyHost(false));

        testConsole(webClient);
    }

    @Test
    @DisabledIfEnvironmentVariable(named = Environment.USE_MINUKUBE_ENV, matches = "true")
    void testOpenshiftCertProvider() throws Exception {
        createTestEnv(new CertSpecBuilder()
                .withProvider(CertProvider.openshift.name())
                .build(),
                false);
        String appNamespace = "certificate-validator-ns";
        boolean testSucceeded = false;
        try {
            SystemtestsKubernetesApps.deployOpenshiftCertValidator(appNamespace, kubernetes);
            OpenshiftCertValidatorApiClient client = new OpenshiftCertValidatorApiClient(kubernetes, SystemtestsKubernetesApps.getOpenshiftCertValidatorEndpoint(appNamespace, kubernetes));

            JsonObject request = new JsonObject();
            request.put("username", user.getUsername());
            request.put("password", user.getPassword());

            Endpoint messagingEndpoint = kubernetes.getEndpoint("messaging-"+AddressSpaceUtils.getAddressSpaceInfraUuid(addressSpace), environment.namespace(), "amqps");
            request.put("messagingHost", messagingEndpoint.getHost());
            request.put("messagingPort", messagingEndpoint.getPort());

            Endpoint mqttEndpoint = kubernetes.getEndpoint("mqtt-"+AddressSpaceUtils.getAddressSpaceInfraUuid(addressSpace), environment.namespace(), "secure-mqtt");
            request.put("mqttHost", mqttEndpoint.getHost());
            request.put("mqttPort", Integer.toString(mqttEndpoint.getPort()));

            Endpoint consoleEndpoint = kubernetes.getEndpoint("console-"+AddressSpaceUtils.getAddressSpaceInfraUuid(addressSpace), environment.namespace(), "https");
            request.put("consoleHost", consoleEndpoint.getHost());
            request.put("consolePort", consoleEndpoint.getPort());

            TimeoutBudget timeout = new TimeoutBudget(5, TimeUnit.MINUTES);
            Exception lastException = null;
            while (!timeout.timeoutExpired()) {
                try {
                    log.info("Making request to openshift-cert-validator {}", request);
                    JsonObject response = client.test(request);
                    if (response.containsKey("error")) {
                        fail("Error testing openshift provider " + response.getString("error"));
                    } else {
                        testSucceeded = true;
                        return;
                    }
                } catch (Exception e) {
                    lastException = e;
                }
                log.debug("next iteration, remaining time: {}", timeout.timeLeft());
                Thread.sleep(5000);
            }
            log.error("Timeout expired");
            if (lastException != null) {
                throw lastException;
            }
        } finally {
            if (!testSucceeded) {
                logCollector.collectLogsOfPodsByLabels(appNamespace,
                        null, Collections.singletonMap("app", SystemtestsKubernetesApps.OPENSHIFT_CERT_VALIDATOR));
            }
            SystemtestsKubernetesApps.deleteOpenshiftCertValidator(appNamespace, kubernetes);
        }
    }

    private void createTestEnv(CertSpec endpointCert) throws Exception {
        createTestEnv(endpointCert, true);
    }

    private void createTestEnv(CertSpec endpointCert, boolean createAddresses) throws Exception {
        addressSpace = AddressSpaceUtils.createAddressSpaceObject("cert-provider-addr-space", AddressSpaceType.STANDARD, AuthenticationServiceType.STANDARD);
        addressSpace.getSpec().setEndpoints(Arrays.asList(
                createEndpointSpec("messaging", "amqps", endpointCert),
                createEndpointSpec("console", "https", endpointCert),
                createEndpointSpec("mqtt", "secure-mqtt", endpointCert)));
        createAddressSpace(addressSpace);

        user = new UserCredentials("user1", "password1");
        createUser(addressSpace, user);

        if (createAddresses) {
            queue = AddressUtils.createQueueAddressObject("test-queue", DestinationPlan.STANDARD_SMALL_QUEUE);
            topic = AddressUtils.createTopicAddressObject("mytopic", DestinationPlan.STANDARD_LARGE_TOPIC);
            setAddresses(addressSpace, queue, topic);
        }
    }

    private EndpointSpec createEndpointSpec(String service, String servicePort, CertSpec endpointCert) {
        return new EndpointSpecBuilder()
                .withName(ENDPOINT_PREFIX + service)
                .withService(service)
                .withExpose(
                new ExposeSpecBuilder()
                .withRouteServicePort(servicePort)
                .withType(ExposeType.route)
                .withRouteTlsTermination(TlsTermination.passthrough)
                .build())
                .withCert(endpointCert)
                .build();
    }

    private void testCertProvider(AmqpClient amqpClient, IMqttClient mqttClient) throws Exception {
        QueueTest.runQueueTest(amqpClient, queue);
        mqttClient.connect();
        simpleMQTTSendReceive(topic, mqttClient, 3);
        mqttClient.disconnect();
    }

    private void testConsole(WebClient webClient) throws Exception {
        Endpoint consoleEndpoint = getConsoleEndpoint(addressSpace);
        CompletableFuture<Optional<Throwable>> promise = new CompletableFuture<>();
        webClient.get(consoleEndpoint.getPort(), consoleEndpoint.getHost(), "").ssl(true).send(ar -> {
            log.info("get console " + ar.toString());
            if (ar.succeeded()) {
                promise.complete(Optional.empty());
            } else {
                log.info("Exception in get console", ar.cause());
                promise.complete(Optional.of(ar.cause()));
            }
        });

        Optional<Throwable> optError = promise.get();
        if (optError.isPresent()) {
            fail(optError.get());
        }
    }

    private SSLSocketFactory getSocketFactory(InputStream caCrtFile) throws Exception {
        CertificateFactory cf = CertificateFactory.getInstance("X.509");

        X509Certificate caCert = (X509Certificate) cf.generateCertificate(caCrtFile);

        KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
        caKs.load(null, null);
        caKs.setCertificateEntry("ca-certificate", caCert);
        TrustManagerFactory tmf = TrustManagerFactory.getInstance("X509");
        tmf.init(caKs);

        SSLContext context = mqttClientFactory.tryGetSSLContext("TLSv1.2", "TLSv1.1", "TLS", "TLSv1");
        context.init(null, tmf.getTrustManagers(), new SecureRandom());

        return context.getSocketFactory();
    }


}
