/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.connect.rest.basic.auth.extension;

import org.apache.kafka.common.security.authenticator.TestJaasConfig;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.test.TestUtils;

import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Map;

import javax.security.auth.callback.Callback;
import javax.security.auth.callback.CallbackHandler;
import javax.security.auth.callback.ChoiceCallback;

import jakarta.ws.rs.HttpMethod;
import jakarta.ws.rs.container.ContainerRequestContext;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.SecurityContext;
import jakarta.ws.rs.core.UriInfo;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;

public class JaasBasicAuthFilterTest {

    private static final String LOGIN_MODULE =
        "org.apache.kafka.connect.rest.basic.auth.extension.PropertyFileLoginModule";

    @Test
    public void testSuccess() throws IOException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testEmptyCredentialsFile() throws IOException {
        File credentialFile = setupPropertyLoginFile(false);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testBadCredential() throws IOException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user1", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testBadPassword() throws IOException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user", "password1");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testUnknownBearer() throws IOException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Unknown", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testUnknownLoginModule() throws IOException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect1", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testUnknownCredentialsFile() {
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", "/tmp/testcrednetial");
        ContainerRequestContext requestContext = setMock("Basic", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testNoFileOption() {
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", null);
        ContainerRequestContext requestContext = setMock("Basic", "user", "password");
        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
    }

    @Test
    public void testInternalTaskConfigEndpointSkipped() throws IOException {
        testInternalEndpointSkipped(HttpMethod.POST, "connectors/connName/tasks");
    }

    @Test
    public void testInternalZombieFencingEndpointSkipped() throws IOException {
        testInternalEndpointSkipped(HttpMethod.PUT, "connectors/connName/fence");
    }

    private void testInternalEndpointSkipped(String method, String endpoint) throws IOException {
        UriInfo uriInfo = mock(UriInfo.class);
        when(uriInfo.getPath()).thenReturn(endpoint);

        ContainerRequestContext requestContext = mock(ContainerRequestContext.class);
        when(requestContext.getMethod()).thenReturn(method);
        when(requestContext.getUriInfo()).thenReturn(uriInfo);

        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect1", credentialFile.getPath());

        jaasBasicAuthFilter.filter(requestContext);

        verify(uriInfo).getPath();
        verify(requestContext, atLeastOnce()).getMethod();
        verify(requestContext).getUriInfo();
        verifyNoMoreInteractions(requestContext);
    }

    @Test
    public void testPostNotChangingConnectorTask() throws IOException {
        UriInfo uriInfo = mock(UriInfo.class);
        when(uriInfo.getPath()).thenReturn("local:randomport/connectors/connName");

        ContainerRequestContext requestContext = mock(ContainerRequestContext.class);
        when(requestContext.getMethod()).thenReturn(HttpMethod.POST);
        when(requestContext.getUriInfo()).thenReturn(uriInfo);
        String authHeader = "Basic" + Base64.getEncoder().encodeToString(("user" + ":" + "password").getBytes());
        when(requestContext.getHeaderString(JaasBasicAuthFilter.AUTHORIZATION))
            .thenReturn(authHeader);

        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());

        jaasBasicAuthFilter.filter(requestContext);

        verify(requestContext).abortWith(any(Response.class));
        verify(requestContext).getUriInfo();
        verify(requestContext).getUriInfo();
    }

    @Test
    public void testUnsupportedCallback() {
        CallbackHandler callbackHandler = new JaasBasicAuthFilter.BasicAuthCallBackHandler(
                new JaasBasicAuthFilter.BasicAuthCredentials(authHeader("basic", "user", "pwd")));
        Callback unsupportedCallback = new ChoiceCallback(
            "You take the blue pill... the story ends, you wake up in your bed and believe whatever you want to believe. " 
                + "You take the red pill... you stay in Wonderland, and I show you how deep the rabbit hole goes.",
            new String[] {"blue pill", "red pill"},
            1,
            true
        );
        assertThrows(ConnectException.class, () -> callbackHandler.handle(new Callback[] {unsupportedCallback}));
    }

    @Test
    public void testSecurityContextSet() throws IOException, URISyntaxException {
        File credentialFile = setupPropertyLoginFile(true);
        JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", credentialFile.getPath());
        ContainerRequestContext requestContext = setMock("Basic", "user1", "password1");

        when(requestContext.getUriInfo()).thenReturn(mock(UriInfo.class));
        when(requestContext.getUriInfo().getRequestUri()).thenReturn(new URI("https://foo.bar"));

        jaasBasicAuthFilter.filter(requestContext);

        ArgumentCaptor<SecurityContext> capturedContext = ArgumentCaptor.forClass(SecurityContext.class);
        verify(requestContext).setSecurityContext(capturedContext.capture());
        assertEquals("user1", capturedContext.getValue().getUserPrincipal().getName());
        assertTrue(capturedContext.getValue().isSecure());
    }

    private String authHeader(String authorization, String username, String password) {
        return authorization + " " + Base64.getEncoder().encodeToString((username + ":" + password).getBytes());
    }

    private ContainerRequestContext setMock(String authorization, String username, String password) {
        ContainerRequestContext requestContext = mock(ContainerRequestContext.class);
        when(requestContext.getMethod()).thenReturn(HttpMethod.GET);
        when(requestContext.getHeaderString(JaasBasicAuthFilter.AUTHORIZATION))
            .thenReturn(authHeader(authorization, username, password));
        return requestContext;
    }

    private File setupPropertyLoginFile(boolean includeUsers) throws IOException {
        File credentialFile = TestUtils.tempFile("credential", ".properties");
        if (includeUsers) {
            List<String> lines = new ArrayList<>();
            lines.add("user=password");
            lines.add("user1=password1");
            Files.write(credentialFile.toPath(), lines, StandardCharsets.UTF_8);
        }
        return credentialFile;
    }

    private JaasBasicAuthFilter setupJaasFilter(String name, String credentialFilePath) {
        TestJaasConfig configuration = new TestJaasConfig();
        Map<String, Object> moduleOptions = credentialFilePath != null
            ? Map.of("file", credentialFilePath)
            : Map.of();
        configuration.addEntry(name, LOGIN_MODULE, moduleOptions);
        return new JaasBasicAuthFilter(configuration);
    }

}
