content: /*
   * 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.ambari.server;
  
  import java.util.concurrent.Callable;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.FutureTask;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  
  import org.apache.ambari.server.configuration.Configuration;
  import org.apache.commons.lang.StringUtils;
  import org.apache.directory.kerberos.client.KdcConfig;
  import org.apache.directory.kerberos.client.KdcConnection;
  import org.apache.directory.shared.kerberos.KerberosMessageType;
  import org.apache.directory.shared.kerberos.exceptions.ErrorType;
  import org.apache.directory.shared.kerberos.exceptions.KerberosException;
  import org.apache.directory.shared.kerberos.messages.KrbError;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.google.inject.Inject;
  import com.google.inject.Singleton;
  
  /**
   * Utility class which checks connection to Kerberos Server.
   * <p>
   * It has two potential clients.
   * <ul>
   * <li>Ambari Agent:
   * Uses it to make sure host can talk to specified KDC Server
   * </li>
   * <p/>
   * <li>Ambari Server:
   * Uses it for connection check, like agent, and also validates
   * the credentials provided on Server side.
   * </li>
   * </ul>
   * </p>
   */
  @Singleton
  public class KdcServerConnectionVerification {
  
    private static final Logger LOG = LoggerFactory.getLogger(KdcServerConnectionVerification.class);
  
    private Configuration config;
  
    /**
     * The connection timeout in seconds.
     */
    private int connectionTimeout = 10;
  
    @Inject
    public KdcServerConnectionVerification(Configuration config) {
      this.config = config;
    }
  
  
    /**
     * Given server IP or hostname, checks if server is reachable i.e.
     * we can make a socket connection to it. Hostname may contain port
     * number separated by a colon.
     *
     * @param kdcHost KDC server IP or hostname (with optional port number)
     * @return true, if server is accepting connection given port; false otherwise.
     */
    public boolean isKdcReachable(String kdcHost) {
      try {
        if (kdcHost == null || kdcHost.isEmpty()) {
          throw new IllegalArgumentException("Invalid hostname for KDC server");
        }
        String[] kdcDetails = kdcHost.split(":");
        if (kdcDetails.length == 1) {
          return isKdcReachable(kdcDetails[0], parsePort(config.getDefaultKdcPort()));
        } else {
          return isKdcReachable(kdcDetails[0], parsePort(kdcDetails[1]));
        }
      } catch (Exception e) {
        LOG.error("Exception while checking KDC reachability: " + e);
        return false;
      }
    }
  
    /**
     * Given a host and port, checks if server is reachable meaning that we
     * can communicate with it.  First we attempt to connect via TCP and if
     * that is unsuccessful, attempt via UDP. It is important to understand that
     * we are not validating credentials, only attempting to communicate with server
     * process for the give host and port.
     *
     * @param server KDC server IP or hostname
     * @param port   KDC port
     * @return true, if server is accepting connection given port; false otherwise.
     */
    public boolean isKdcReachable(String server, int port) {
      boolean success = isKdcReachable(server, port, ConnectionProtocol.TCP) || isKdcReachable(server, port, ConnectionProtocol.UDP);
  
      if (!success) {
        LOG.error("Failed to connect to the KDC at {}:{} using either TCP or UDP", server, port);
      }
  
      return success;
    }
  
    /**
     * Attempt to communicate with KDC server over a specified communication protocol (TCP or UDP).
     *
     * @param server         KDC hostname or IP address
     * @param port           KDC server port
     * @param connectionProtocol the type of connection to use
     * @return true if communication is successful; false otherwise
     */
    public boolean isKdcReachable(final String server, final int port, final ConnectionProtocol connectionProtocol) {
      int timeoutMillis = connectionTimeout * 1000;
      final KdcConfig config = KdcConfig.getDefaultConfig();
      config.setHostName(server);
      config.setKdcPort(port);
      config.setUseUdp(ConnectionProtocol.UDP == connectionProtocol);
      config.setTimeout(timeoutMillis);
  
      FutureTask<Boolean> future = new FutureTask<>(new Callable<Boolean>() {
        @Override
        public Boolean call() {
          Boolean success;
  
          try {
            KdcConnection connection = getKdcConnection(config);
            // we are only testing whether we can communicate with server and not
            // validating credentials
            connection.getTgt("noUser@noRealm", "noPassword");
  
            LOG.info(String.format("Encountered no Exceptions while testing connectivity to the KDC:\n" +
                "**** Host: %s:%d (%s)",
              server, port, connectionProtocol.name()));
            success = true;
          } catch (KerberosException e) {
            KrbError error = e.getError();
            ErrorType errorCode = error.getErrorCode();
  
            String errorCodeMessage;
            int errorCodeCode;
            if (errorCode != null) {
              errorCodeMessage = errorCode.getMessage();
              errorCodeCode = errorCode.getValue();
            } else {
              errorCodeMessage = "<Not Specified>";
              errorCodeCode = -1;
            }
  
            // unfortunately, need to look at msg as error 60 is a generic error code
            //todo: evaluate other error codes to provide better information
            //todo: as there may be other error codes where we should return false
            success = !(errorCodeCode == ErrorType.KRB_ERR_GENERIC.getValue() &&
              errorCodeMessage.contains("TimeOut"));
  
            if (!success || LOG.isDebugEnabled()) {
              KerberosMessageType messageType = error.getMessageType();
  
              String messageTypeMessage;
              int messageTypeCode;
              if (messageType != null) {
                messageTypeMessage = messageType.getMessage();
                messageTypeCode = messageType.getValue();
              } else {
                messageTypeMessage = "<Not Specified>";
                messageTypeCode = -1;
              }
  
              String message = String.format("Received KerberosException while testing connectivity to the KDC: %s\n" +
                  "**** Host:    %s:%d (%s)\n" +
                  "**** Error:   %s\n" +
                  "**** Code:    %d (%s)\n" +
                  "**** Message: %d (%s)",
                e.getLocalizedMessage(), server, port, connectionProtocol.name(), error.getEText(), errorCodeCode,
                errorCodeMessage, messageTypeCode, messageTypeMessage);
  
              if (LOG.isDebugEnabled()) {
                LOG.info(message, e);
              } else {
                LOG.info(message);
              }
            }
          } catch (Throwable e) {
            LOG.info(String.format("Received Exception while testing connectivity to the KDC: %s\n**** Host: %s:%d (%s)",
              e.getLocalizedMessage(), server, port, connectionProtocol.name()), e);
  
            // some bad unexpected thing occurred
            throw new RuntimeException(e);
          }
  
          return success;
        }
      });
  
      new Thread(future, "ambari-kdc-verify").start();
      Boolean result;
      try {
        // timeout after specified timeout
        result = future.get(timeoutMillis, TimeUnit.MILLISECONDS);
  
        if (result) {
          LOG.info(String.format("Successfully connected to the KDC server at %s:%d over %s",
              server, port, connectionProtocol.name()));
        } else {
          LOG.warn(String.format("Failed to connect to the KDC server at %s:%d over %s",
              server, port, connectionProtocol.name()));
        }
      } catch (InterruptedException e) {
        String message = String.format("Interrupted while trying to communicate with KDC server at %s:%d over %s",
            server, port, connectionProtocol.name());
        if (LOG.isDebugEnabled()) {
          LOG.warn(message, e);
        } else {
          LOG.warn(message);
        }
  
        result = false;
        future.cancel(true);
      } catch (ExecutionException e) {
        String message = String.format("An unexpected exception occurred while attempting to communicate with the KDC server at %s:%d over %s",
            server, port, connectionProtocol.name());
        if (LOG.isDebugEnabled()) {
          LOG.warn(message, e);
        } else {
          LOG.warn(message);
        }
  
        result = false;
      } catch (TimeoutException e) {
        String message = String.format("Timeout occurred while attempting to to communicate with KDC server at %s:%d over %s",
            server, port, connectionProtocol.name());
        if (LOG.isDebugEnabled()) {
          LOG.warn(message, e);
        } else {
          LOG.warn(message);
        }
  
        result = false;
        future.cancel(true);
      }
  
      return result;
    }
  
    /**
     * Get a KDC UDP connection for the given configuration.
     * This has been extracted into it's own method primarily
     * for unit testing purposes.
     *
     * @param config KDC connection configuration
     * @return new KDC connection
     */
    protected KdcConnection getKdcConnection(KdcConfig config) {
      return new KdcConnection(config);
    }
  
    /**
     * Set the connection timeout.
     * This is the amount of time that we will attempt to read data from connection.
     *
     * @param timeoutSeconds timeout in seconds
     */
    public void setConnectionTimeout(int timeoutSeconds) {
      connectionTimeout = (timeoutSeconds < 1) ? 1 : timeoutSeconds;
    }
  
    /**
     * Get the timeout value.
     *
     * @return the connection timeout value in seconds
     */
    public int getConnectionTimeout() {
      return connectionTimeout;
    }
  
    /**
     * Parses port number from given string.
     *
     * @param port port number string
     * @return parsed port number
     * @throws NumberFormatException    if given string cannot be parsed
     * @throws IllegalArgumentException if given string is null or empty
     */
    protected int parsePort(String port) {
      if (StringUtils.isEmpty(port)) {
        throw new IllegalArgumentException("Port number must be non-empty, non-null positive integer");
      }
      return Integer.parseInt(port);
    }
  
    /**
     * A connection protocol to use to for connecting to the KDC
     */
    public enum ConnectionProtocol {
      TCP,
      UDP
    }
  }
  