HDDS-1060. Add API to get OM certificate from SCM CA. Contributed by Ajay Kumar.

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/HddsClientUtils.java

@@ -22,15 +22,29 @@ import com.google.common.base.Preconditions;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.HddsUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolClientSideTranslatorPB;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
+import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolPB;
+import org.apache.hadoop.ipc.Client;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.text.ParseException;
 import java.time.Instant;
 import java.time.ZoneId;
@@ -38,6 +52,7 @@ import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.concurrent.TimeUnit;
 
+
 /**
  * Utility methods for Ozone and Container Clients.
  *
@@ -252,4 +267,27 @@ public final class HddsClientUtils {
             ScmConfigKeys
                 .SCM_CONTAINER_CLIENT_MAX_OUTSTANDING_REQUESTS_DEFAULT);
   }
+
+  /**
+   * Create a scm block client, used by putKey() and getKey().
+   *
+   * @return {@link ScmBlockLocationProtocol}
+   * @throws IOException
+   */
+  public static SCMSecurityProtocol getScmSecurityClient(
+      OzoneConfiguration conf, UserGroupInformation ugi) throws IOException {
+    RPC.setProtocolEngine(conf, SCMSecurityProtocolPB.class,
+        ProtobufRpcEngine.class);
+    long scmVersion =
+        RPC.getProtocolVersion(ScmBlockLocationProtocolPB.class);
+    InetSocketAddress scmSecurityProtoAdd =
+        HddsUtils.getScmAddressForSecurityProtocol(conf);
+    SCMSecurityProtocolClientSideTranslatorPB scmSecurityClient =
+        new SCMSecurityProtocolClientSideTranslatorPB(
+            RPC.getProxy(SCMSecurityProtocolPB.class, scmVersion,
+                scmSecurityProtoAdd, ugi, conf,
+                NetUtils.getDefaultSocketFactory(conf),
+                Client.getRpcTimeout(conf)));
+    return scmSecurityClient;
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java

@@ -410,4 +410,54 @@ public final class HddsUtils {
   public static long getUtcTime() {
     return Calendar.getInstance(UTC_ZONE).getTimeInMillis();
   }
+
+  /**
+   * Retrieve the socket address that should be used by clients to connect
+   * to the SCM for
+   * {@link org.apache.hadoop.hdds.protocol.SCMSecurityProtocol}. If
+   * {@link ScmConfigKeys#OZONE_SCM_SECURITY_SERVICE_ADDRESS_KEY} is not defined
+   * then {@link ScmConfigKeys#OZONE_SCM_CLIENT_ADDRESS_KEY} is used. If neither
+   * is defined then {@link ScmConfigKeys#OZONE_SCM_NAMES} is used.
+   *
+   * @param conf
+   * @return Target InetSocketAddress for the SCM block client endpoint.
+   * @throws IllegalArgumentException if configuration is not defined.
+   */
+  public static InetSocketAddress getScmAddressForSecurityProtocol(
+      Configuration conf) {
+    Optional<String> host = getHostNameFromConfigKeys(conf,
+        ScmConfigKeys.OZONE_SCM_SECURITY_SERVICE_ADDRESS_KEY);
+
+    if (!host.isPresent()) {
+      host = getHostNameFromConfigKeys(conf,
+          ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);
+    }
+
+    if (!host.isPresent()) {
+      // Fallback to Ozone SCM names.
+      Collection<InetSocketAddress> scmAddresses = getSCMAddresses(conf);
+      if (scmAddresses.size() > 1) {
+        throw new IllegalArgumentException(
+            ScmConfigKeys.OZONE_SCM_NAMES +
+                " must contain a single hostname. Multiple SCM hosts are " +
+                "currently unsupported");
+      }
+      host = Optional.of(scmAddresses.iterator().next().getHostName());
+    }
+
+    if (!host.isPresent()) {
+      throw new IllegalArgumentException(
+          ScmConfigKeys.OZONE_SCM_SECURITY_SERVICE_ADDRESS_KEY
+              + " must be defined. See"
+              + " https://wiki.apache.org/hadoop/Ozone#Configuration"
+              + " for details on configuring Ozone.");
+    }
+
+    final Optional<Integer> port = getPortNumberFromConfigKeys(conf,
+        ScmConfigKeys.OZONE_SCM_SECURITY_SERVICE_PORT_KEY);
+
+    return NetUtils.createSocketAddr(host.get() + ":" + port
+        .orElse(ScmConfigKeys.OZONE_SCM_SECURITY_SERVICE_PORT_DEFAULT));
+  }
+
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/SCMSecurityProtocol.java

@@ -47,9 +47,28 @@ public interface SCMSecurityProtocol {
    *
    * @param omDetails       - DataNode Details.
    * @param certSignReq     - Certificate signing request.
-   * @return byte[]         - SCM signed certificate.
+   * @return String         - pem encoded SCM signed
+   *                          certificate.
    */
   String getOMCertificate(OzoneManagerDetailsProto omDetails,
       String certSignReq) throws IOException;
 
+  /**
+   * Get SCM signed certificate for given certificate serial id if it exists.
+   * Throws exception if it's not found.
+   *
+   * @param certSerialId    - Certificate serial id.
+   * @return String         - pem encoded SCM signed
+   *                          certificate with given cert id if it
+   *                          exists.
+   */
+  String getCertificate(String certSerialId) throws IOException;
+
+  /**
+   * Get CA certificate.
+   *
+   * @return String         - pem encoded CA certificate.
+   */
+  String getCACertificate() throws IOException;
+
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocolPB/SCMSecurityProtocolClientSideTranslatorPB.java

@@ -22,6 +22,9 @@ import java.io.Closeable;
 import java.io.IOException;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.OzoneManagerDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCACertificateRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertificateRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertificateRequestProto.Builder;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetDataNodeCertRequestProto;
 import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
 import org.apache.hadoop.ipc.ProtobufHelper;
@@ -112,6 +115,43 @@ public class SCMSecurityProtocolClientSideTranslatorPB implements
     }
   }
 
+  /**
+   * Get SCM signed certificate with given serial id. Throws exception if
+   * certificate is not found.
+   *
+   * @param certSerialId    - Certificate serial id.
+   * @return string         - pem encoded certificate.
+   */
+  @Override
+  public String getCertificate(String certSerialId) throws IOException {
+    Builder builder = SCMGetCertificateRequestProto
+        .newBuilder()
+        .setCertSerialId(certSerialId);
+    try {
+      return rpcProxy.getCertificate(NULL_RPC_CONTROLLER, builder.build())
+          .getX509Certificate();
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
+  /**
+   * Get CA certificate.
+   *
+   * @return serial   - Root certificate.
+   */
+  @Override
+  public String getCACertificate() throws IOException {
+    SCMGetCACertificateRequestProto protoIns = SCMGetCACertificateRequestProto
+        .getDefaultInstance();
+    try {
+      return rpcProxy.getCACertificate(NULL_RPC_CONTROLLER, protoIns)
+          .getX509Certificate();
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+  }
+
   /**
    * Return the proxy object underlying this protocol translator.
    *

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocolPB/SCMSecurityProtocolServerSideTranslatorPB.java

@@ -20,7 +20,9 @@ import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
 import java.io.IOException;
 
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertificateRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetDataNodeCertRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertResponseProto.ResponseCode;
 import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
@@ -91,4 +93,37 @@ public class SCMSecurityProtocolServerSideTranslatorPB implements
       throw new ServiceException(e);
     }
   }
+
+  @Override
+  public SCMGetCertResponseProto getCertificate(RpcController controller,
+      SCMGetCertificateRequestProto request) throws ServiceException {
+    try {
+      String certificate = impl.getCertificate(request.getCertSerialId());
+      SCMGetCertResponseProto.Builder builder =
+          SCMGetCertResponseProto
+              .newBuilder()
+              .setResponseCode(ResponseCode.success)
+              .setX509Certificate(certificate);
+      return builder.build();
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
+
+  @Override
+  public SCMGetCertResponseProto getCACertificate(RpcController controller,
+      SCMSecurityProtocolProtos.SCMGetCACertificateRequestProto request)
+      throws ServiceException {
+    try {
+      String certificate = impl.getCACertificate();
+      SCMGetCertResponseProto.Builder builder =
+          SCMGetCertResponseProto
+              .newBuilder()
+              .setResponseCode(ResponseCode.success)
+              .setX509Certificate(certificate);
+      return builder.build();
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateServer.java

@@ -21,6 +21,7 @@ package org.apache.hadoop.hdds.security.x509.certificate.authority;
 
 import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
 import org.apache.hadoop.hdds.security.x509.SecurityConfig;
+import org.apache.hadoop.hdds.security.x509.certificate.authority.CertificateApprover.ApprovalType;
 import org.bouncycastle.cert.X509CertificateHolder;
 import org.bouncycastle.pkcs.PKCS10CertificationRequest;
 
@@ -30,8 +31,8 @@ import java.security.cert.X509Certificate;
 import java.util.concurrent.Future;
 
 /**
- * Interface for Certificate Authority. This can be extended to talk to external
+ * Interface for Certificate Authority. This can be extended to talk to
- * CAs later or HSMs later.
+ * external CAs later or HSMs later.
  */
 public interface CertificateServer {
   /**
@@ -56,6 +57,18 @@ public interface CertificateServer {
   X509CertificateHolder getCACertificate()
       throws CertificateException, IOException;
 
+  /**
+   * Returns the Certificate corresponding to given certificate serial id if
+   * exist. Return null if it doesn't exist.
+   *
+   * @return certSerialId         - Certificate serial id.
+   * @throws CertificateException - usually thrown if this CA is not
+   *                              initialized.
+   * @throws IOException          - on Error.
+   */
+  X509Certificate getCertificate(String certSerialId)
+      throws CertificateException, IOException;
+
   /**
    * Request a Certificate based on Certificate Signing Request.
    *
@@ -80,11 +93,8 @@ public interface CertificateServer {
    * approved.
    * @throws SCMSecurityException - on Error.
    */
-  Future<X509CertificateHolder>
+  Future<X509CertificateHolder> requestCertificate(String csr,
-      requestCertificate(String csr, CertificateApprover.ApprovalType type)
+      ApprovalType type) throws IOException;
-      throws IOException;
-
-
 
   /**
    * Revokes a Certificate issued by this CertificateServer.
@@ -95,7 +105,7 @@ public interface CertificateServer {
    * @throws SCMSecurityException - on Error.
    */
   Future<Boolean> revokeCertificate(X509Certificate certificate,
-      CertificateApprover.ApprovalType approver) throws SCMSecurityException;
+      ApprovalType approver) throws SCMSecurityException;
 
   /**
    * TODO : CRL, OCSP etc. Later. This is the start of a CertificateServer

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/DefaultCAServer.java

@@ -36,6 +36,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.math.BigInteger;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -171,6 +172,23 @@ public class DefaultCAServer implements CertificateServer {
     }
   }
 
+  /**
+   * Returns the Certificate corresponding to given certificate serial id if
+   * exist. Return null if it doesn't exist.
+   *
+   * @param certSerialId         - Certificate for this CA.
+   * @return X509CertificateHolder
+   * @throws CertificateException - usually thrown if this CA is not
+   * initialized.
+   * @throws IOException - on Error.
+   */
+  @Override
+  public X509Certificate getCertificate(String certSerialId) throws
+      IOException {
+    return store.getCertificateByID(new BigInteger(certSerialId),
+        CertificateStore.CertType.VALID_CERTS);
+  }
+
   private KeyPair getCAKeys() throws IOException {
     KeyCodec keyCodec = new KeyCodec(config, componentName);
     try {

hadoop-hdds/common/src/main/proto/SCMSecurityProtocol.proto

@@ -52,6 +52,19 @@ message SCMGetOMCertRequestProto {
   required string CSR = 2;
 }
 
+/**
+* Proto request to get a certificate with given serial id.
+*/
+message SCMGetCertificateRequestProto {
+  required string certSerialId = 1;
+}
+
+/**
+* Proto request to get CA certificate.
+*/
+message SCMGetCACertificateRequestProto {
+}
+
 /**
  * Returns a certificate signed by SCM.
  */
@@ -79,4 +92,15 @@ service SCMSecurityProtocolService {
   rpc getOMCertificate (SCMGetOMCertRequestProto) returns
   (SCMGetCertResponseProto);
 
+  /**
+   * Get SCM signed certificate for DataNode.
+   */
+  rpc getCertificate (SCMGetCertificateRequestProto) returns
+  (SCMGetCertResponseProto);
+
+  /**
+   * Get SCM signed certificate for DataNode.
+   */
+  rpc getCACertificate (SCMGetCACertificateRequestProto) returns
+  (SCMGetCertResponseProto);
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMSecurityProtocolServer.java

@@ -19,6 +19,8 @@ package org.apache.hadoop.hdds.scm.server;
 import com.google.protobuf.BlockingService;
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -137,6 +139,47 @@ public class SCMSecurityProtocolServer implements SCMSecurityProtocol {
     }
   }
 
+  /**
+   * Get SCM signed certificate with given serial id.
+   *
+   * @param certSerialId    - Certificate serial id.
+   * @return string         - pem encoded SCM signed certificate.
+   */
+  @Override
+  public String getCertificate(String certSerialId) throws IOException {
+    LOGGER.debug("Getting certificate with certificate serial id",
+        certSerialId);
+    try {
+      X509Certificate certificate =
+          certificateServer.getCertificate(certSerialId);
+      if (certificate != null) {
+        return CertificateCodec.getPEMEncodedString(certificate);
+      }
+    } catch (CertificateException e) {
+      LOGGER.error("getCertificate operation failed. ", e);
+      throw new IOException("getCertificate operation failed. ", e);
+    }
+    LOGGER.debug("Certificate with serial id {} not found.", certSerialId);
+    throw new IOException("Certificate not found");
+  }
+
+  /**
+   * Get SCM signed certificate for OM.
+   *
+   * @return string         - Root certificate.
+   */
+  @Override
+  public String getCACertificate() throws IOException {
+    LOGGER.debug("Getting CA certificate.");
+    try {
+      return CertificateCodec.getPEMEncodedString(
+          certificateServer.getCACertificate());
+    } catch (CertificateException e) {
+      LOGGER.error("getRootCertificate operation failed. ", e);
+      throw new IOException("getRootCertificate operation failed. ", e);
+    }
+  }
+
   public RPC.Server getRpcServer() {
     return rpcServer;
   }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestSecureOzoneCluster.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.ozone;
 
+import static junit.framework.TestCase.assertNotNull;
 import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_METADATA_DIRS;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ADMINISTRATORS;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ENABLED;
@@ -25,6 +26,7 @@ import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVA
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.TOKEN_ERROR_OTHER;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.TOKEN_EXPIRED;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.VOLUME_NOT_FOUND;
+import static org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod.KERBEROS;
 import static org.slf4j.event.Level.INFO;
 
 import java.io.File;
@@ -37,14 +39,17 @@ import java.security.PrivilegedExceptionAction;
 import java.util.Properties;
 import java.util.UUID;
 import java.util.concurrent.Callable;
+
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
 import org.apache.hadoop.hdds.scm.server.SCMStorageConfig;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
@@ -53,6 +58,7 @@ import org.apache.hadoop.hdds.security.x509.keys.KeyCodec;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.ipc.Client;
 import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ipc.Server;
 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.minikdc.MiniKdc;
@@ -105,7 +111,10 @@ public final class TestSecureOzoneCluster {
   private File scmKeytab;
   private File spnegoKeytab;
   private File omKeyTab;
+  private File testUserKeytab;
   private String curUser;
+  private String testUserPrincipal;
+  private UserGroupInformation testKerberosUgi;
   private StorageContainerManager scm;
   private OzoneManager om;
 
@@ -163,8 +172,7 @@ public final class TestSecureOzoneCluster {
     createPrincipal(spnegoKeytab,
         configuration.get(ScmConfigKeys
             .HDDS_SCM_HTTP_KERBEROS_PRINCIPAL_KEY));
-    configuration.get(OMConfigKeys
+    createPrincipal(testUserKeytab, testUserPrincipal);
-        .OZONE_OM_HTTP_KERBEROS_PRINCIPAL_KEY);
     createPrincipal(omKeyTab,
         configuration.get(OMConfigKeys.OZONE_OM_KERBEROS_PRINCIPAL_KEY));
   }
@@ -212,6 +220,8 @@ public final class TestSecureOzoneCluster {
     scmKeytab = new File(workDir, "scm.keytab");
     spnegoKeytab = new File(workDir, "http.keytab");
     omKeyTab = new File(workDir, "om.keytab");
+    testUserKeytab = new File(workDir, "testuser.keytab");
+    testUserPrincipal = "test@" + realm;
 
     configuration.set(ScmConfigKeys.HDDS_SCM_KERBEROS_KEYTAB_FILE_KEY,
         scmKeytab.getAbsolutePath());
@@ -235,6 +245,47 @@ public final class TestSecureOzoneCluster {
     Assert.assertEquals(scmId, scmInfo.getScmId());
   }
 
+  @Test
+  public void testSCMSecurityProtocol() throws Exception {
+
+    initSCM();
+    scm = StorageContainerManager.createSCM(null, conf);
+    //Reads the SCM Info from SCM instance
+    try {
+      scm.start();
+
+      // Case 1: User with Kerberos credentials should succeed.
+      UserGroupInformation ugi =
+          UserGroupInformation.loginUserFromKeytabAndReturnUGI(
+              testUserPrincipal, testUserKeytab.getCanonicalPath());
+      ugi.setAuthenticationMethod(KERBEROS);
+      SCMSecurityProtocol scmSecurityProtocolClient =
+          HddsClientUtils.getScmSecurityClient(conf, ugi);
+      assertNotNull(scmSecurityProtocolClient);
+      String caCert = scmSecurityProtocolClient.getCACertificate();
+      LambdaTestUtils.intercept(RemoteException.class, "Certificate not found",
+          () -> scmSecurityProtocolClient.getCertificate("1"));
+      assertNotNull(caCert);
+
+      // Case 2: User without Kerberos credentials should fail.
+      ugi = UserGroupInformation.createRemoteUser("test");
+      ugi.setAuthenticationMethod(AuthMethod.TOKEN);
+      SCMSecurityProtocol finalScmSecurityProtocolClient =
+          HddsClientUtils.getScmSecurityClient(conf, ugi);
+
+      LambdaTestUtils.intercept(IOException.class, "Client cannot" +
+              " authenticate via:[KERBEROS]",
+          () -> finalScmSecurityProtocolClient.getCACertificate());
+      LambdaTestUtils.intercept(IOException.class, "Client cannot" +
+              " authenticate via:[KERBEROS]",
+          () -> finalScmSecurityProtocolClient.getCertificate("1"));
+    } finally {
+      if (scm != null) {
+        scm.stop();
+      }
+    }
+  }
+
   private void initSCM()
       throws IOException, AuthenticationException {