/*
 * 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.qpid.proton4j.engine.sasl.client;

import java.util.Objects;

import javax.security.sasl.SaslException;

import org.apache.qpid.proton4j.amqp.Symbol;
import org.apache.qpid.proton4j.buffer.ProtonBuffer;
import org.apache.qpid.proton4j.common.logging.ProtonLogger;
import org.apache.qpid.proton4j.common.logging.ProtonLoggerFactory;
import org.apache.qpid.proton4j.engine.sasl.SaslClientContext;
import org.apache.qpid.proton4j.engine.sasl.SaslClientListener;
import org.apache.qpid.proton4j.engine.sasl.SaslOutcome;

/**
 * Handles SASL traffic from the proton engine and drives the authentication process
 */
public class SaslAuthenticator implements SaslClientListener {

    private static final ProtonLogger LOG = ProtonLoggerFactory.getLogger(SaslAuthenticator.class);

    private final SaslMechanismSelector selector;
    private final SaslCredentialsProvider credentials;

    private Mechanism chosenMechanism;

    /**
     * Creates a new SASL Authenticator initialized with the given credentials provider instance.  Because no
     * {@link Mechanism} selector is given the full set of supported SASL mechanisms will be chosen from when
     * attempting to match one to the server offered SASL mechanisms.
     *
     * @param credentials
     *      The credentials that will be used when the SASL negotiation is in progress.
     */
    public SaslAuthenticator(SaslCredentialsProvider credentials) {
        this(new SaslMechanismSelector(), credentials);
    }

    /**
     * Creates a new client SASL Authenticator with the given {@link Mechanism} and client credentials
     * provider instances.  The configured {@link Mechanism} selector is used when attempting to match
     * a SASL {@link Mechanism} with the server offered set of supported SASL mechanisms.
     *
     * @param selector
     *     The {@link SaslMechanismSelector} that will be called upon to choose a server supported mechanism.
     * @param credentials
     *      The credentials that will be used when the SASL negotiation is in progress.
     */
    public SaslAuthenticator(SaslMechanismSelector selector, SaslCredentialsProvider credentials) {
        Objects.requireNonNull(selector, "A SASL Mechanism selector implementation is required");
        Objects.requireNonNull(credentials, "A SASL Credentials provider implementation is required");

        this.credentials = credentials;
        this.selector = selector;
    }

    @Override
    public void handleSaslMechanisms(SaslClientContext context, Symbol[] mechanisms) {
        chosenMechanism = selector.select(mechanisms, credentials);

        if (chosenMechanism == null) {
            context.saslFailure(new SaslException("Could not find a matching mechanism to begin SASL Negotiations"));
            return;
        }

        LOG.debug("SASL Negotiations proceeding using selected mechanisms: {}", chosenMechanism);

        ProtonBuffer initialResponse = null;
        try {
            initialResponse = chosenMechanism.getInitialResponse(credentials);
        } catch (SaslException se) {
            context.saslFailure(se);
        } catch (Throwable unknown) {
            context.saslFailure(new SaslException("Unknown error while fetching initial response", unknown));
        }

        context.sendChosenMechanism(chosenMechanism.getName(), credentials.vhost(), initialResponse);
    }

    @Override
    public void handleSaslChallenge(SaslClientContext context, ProtonBuffer challenge) {
        ProtonBuffer response = null;
        try {
            response = chosenMechanism.getChallengeResponse(credentials, challenge);
        } catch (SaslException se) {
            context.saslFailure(se);
        } catch (Throwable unknown) {
            context.saslFailure(new SaslException("Unknown error while fetching challenge response", unknown));
        }

        context.sendResponse(response);
    }

    @Override
    public void handleSaslOutcome(SaslClientContext context, SaslOutcome outcome, ProtonBuffer additional) {
        try {
            chosenMechanism.verifyCompletion();
        } catch (SaslException se) {
            context.saslFailure(se);
        } catch (Throwable unknown) {
            context.saslFailure(new SaslException("Unknown error while verifying SASL negotiations completion", unknown));
        }
    }
}

@@ -1147,16 +1147,69 @@
 actory;%0A
+import org.apache.qpid.proton4j.engine.EventHandler;%0A
 import o
@@ -1263,16 +1263,16 @@
 ontext;%0A
-
 import o
@@ -1749,16 +1749,75 @@
 tials;%0A%0A
+    private EventHandler%3CSaslOutcome%3E saslCompleteHandler;%0A
     priv
@@ -3347,32 +3347,217 @@
 elector;%0A    %7D%0A%0A
+    public SaslAuthenticator saslComplete(EventHandler%3CSaslOutcome%3E saslCompleteEventHandler) %7B%0A        this.saslCompleteHandler = saslCompleteEventHandler;%0A        return this;%0A    %7D%0A%0A
     @Override%0A  
@@ -5184,16 +5184,130 @@
 tion();%0A
+            if (saslCompleteHandler != null) %7B%0A                saslCompleteHandler.handle(outcome);%0A            %7D%0A
         
@@ -5385,33 +5385,40 @@
    %7D catch (
-Throwable
+RuntimeException
  unknown) %7B%0A

Allow external entity to be signaled when sasl completes