HDDS-980. Adding getOMCertificate in SCMSecurityProtocol. Contributed by Ajay Kumar.

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

@@ -19,6 +19,7 @@ package org.apache.hadoop.hdds.protocol;
 import java.io.IOException;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.OzoneManagerDetailsProto;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.security.KerberosInfo;
 
@@ -41,4 +42,14 @@ public interface SCMSecurityProtocol {
       DatanodeDetailsProto dataNodeDetails,
       String certSignReq) throws IOException;
 
+  /**
+   * Get SCM signed certificate for OM.
+   *
+   * @param omDetails - DataNode Details.
+   * @param certSignReq     - Certificate signing request.
+   * @return byte[]         - SCM signed certificate.
+   */
+  String getOMCertificate(OzoneManagerDetailsProto omDetails,
+      String certSignReq) throws IOException;
+
 }

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

@@ -21,12 +21,15 @@ import com.google.protobuf.ServiceException;
 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.SCMGetDataNodeCertRequestProto;
 import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
 import org.apache.hadoop.ipc.ProtobufHelper;
 import org.apache.hadoop.ipc.ProtocolTranslator;
 import org.apache.hadoop.ipc.RPC;
 
+import static org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetOMCertRequestProto;
+
 /**
  * This class is the client-side translator that forwards requests for
  * {@link SCMSecurityProtocol} to the {@link SCMSecurityProtocolPB} proxy.
@@ -87,6 +90,28 @@ public class SCMSecurityProtocolClientSideTranslatorPB implements
     }
   }
 
+  /**
+   * Get SCM signed certificate for OM.
+   *
+   * @param omDetails       - OzoneManager Details.
+   * @param certSignReq     - Certificate signing request.
+   * @return byte[]         - SCM signed certificate.
+   */
+  @Override
+  public String getOMCertificate(OzoneManagerDetailsProto omDetails,
+      String certSignReq) throws IOException {
+    SCMGetOMCertRequestProto.Builder builder = SCMGetOMCertRequestProto
+        .newBuilder()
+        .setCSR(certSignReq)
+        .setOmDetails(omDetails);
+    try {
+      return rpcProxy.getOMCertificate(NULL_RPC_CONTROLLER, builder.build())
+          .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

@@ -19,10 +19,12 @@ package org.apache.hadoop.hdds.protocolPB;
 import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
 import java.io.IOException;
+
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetDataNodeCertRequestProto;
-import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetDataNodeCertResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertResponseProto.ResponseCode;
-import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetDataNodeCertResponseProto.ResponseCode;
 import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
+import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetOMCertRequestProto;
 
 /**
  * This class is the server-side translator that forwards requests received on
@@ -46,15 +48,41 @@ public class SCMSecurityProtocolServerSideTranslatorPB implements
    * @return SCMGetDataNodeCertResponseProto.
    */
   @Override
-  public SCMGetDataNodeCertResponseProto getDataNodeCertificate(
+  public SCMGetCertResponseProto getDataNodeCertificate(
       RpcController controller, SCMGetDataNodeCertRequestProto request)
       throws ServiceException {
     try {
       String certificate = impl
           .getDataNodeCertificate(request.getDatanodeDetails(),
               request.getCSR());
-      SCMGetDataNodeCertResponseProto.Builder builder =
+      SCMGetCertResponseProto.Builder builder =
-          SCMGetDataNodeCertResponseProto
+          SCMGetCertResponseProto
+              .newBuilder()
+              .setResponseCode(ResponseCode.success)
+              .setX509Certificate(certificate);
+      return builder.build();
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
+
+  /**
+   * Get SCM signed certificate for OzoneManager.
+   *
+   * @param controller
+   * @param request
+   * @return SCMGetCertResponseProto.
+   */
+  @Override
+  public SCMGetCertResponseProto getOMCertificate(
+      RpcController controller, SCMGetOMCertRequestProto request)
+      throws ServiceException {
+    try {
+      String certificate = impl
+          .getOMCertificate(request.getOmDetails(),
+              request.getCSR());
+      SCMGetCertResponseProto.Builder builder =
+          SCMGetCertResponseProto
               .newBuilder()
               .setResponseCode(ResponseCode.success)
               .setX509Certificate(certificate);

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

@@ -32,7 +32,7 @@ import java.util.concurrent.CompletableFuture;
 /**
  * Certificate Approver interface is used to inspectCSR a certificate.
  */
-interface CertificateApprover {
+public interface CertificateApprover {
   /**
    * Approves a Certificate Request based on the policies of this approver.
    *

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/utils/CertificateCodec.java

@@ -119,7 +119,7 @@ public class CertificateCodec {
    * @return PEM Encoded Certificate String.
    * @throws SCMSecurityException - On failure to create a PEM String.
    */
-  public String getPEMEncodedString(X509CertificateHolder x509CertHolder)
+  public static String getPEMEncodedString(X509CertificateHolder x509CertHolder)
       throws SCMSecurityException {
     try {
       StringWriter stringWriter = new StringWriter();

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

@@ -34,10 +34,28 @@ package hadoop.hdds;
 
 import "hdds.proto";
 
+/**
+* This message is send by data node to prove its identity and get an SCM
+* signed certificate.
+*/
+message SCMGetDataNodeCertRequestProto {
+  required DatanodeDetailsProto datanodeDetails = 1;
+  required string CSR = 2;
+}
+
+/**
+* This message is send by OzoneManager to prove its identity and get an SCM
+* signed certificate.
+*/
+message SCMGetOMCertRequestProto {
+  required OzoneManagerDetailsProto omDetails = 1;
+  required string CSR = 2;
+}
+
 /**
  * Returns a certificate signed by SCM.
  */
-message SCMGetDataNodeCertResponseProto {
+message SCMGetCertResponseProto {
   enum ResponseCode {
     success = 1;
     authenticationFailed = 2;
@@ -47,20 +65,18 @@ message SCMGetDataNodeCertResponseProto {
   required string x509Certificate = 2; // Base64 encoded X509 certificate.
 }
 
-/**
-* This message is send by data node to prove its identity and get an SCM
-* signed certificate.
-*/
-message SCMGetDataNodeCertRequestProto {
-  required DatanodeDetailsProto datanodeDetails = 1;
-  required string CSR = 2;
-}
-
 
 service SCMSecurityProtocolService {
   /**
   * Get SCM signed certificate for DataNode.
   */
-  rpc getDataNodeCertificate (SCMGetDataNodeCertRequestProto) returns (SCMGetDataNodeCertResponseProto);
+  rpc getDataNodeCertificate (SCMGetDataNodeCertRequestProto) returns
+  (SCMGetCertResponseProto);
+
+  /**
+  * Get SCM signed certificate for DataNode.
+  */
+  rpc getOMCertificate (SCMGetOMCertRequestProto) returns
+  (SCMGetCertResponseProto);
 
 }

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

@@ -35,6 +35,17 @@ message DatanodeDetailsProto {
     repeated Port ports = 4;
 }
 
+/**
+ Proto message encapsulating information required to uniquely identify a
+ OzoneManager.
+*/
+message OzoneManagerDetailsProto {
+    required string uuid = 1;          // UUID assigned to the OzoneManager.
+    required string ipAddress = 2;     // IP address of OM.
+    required string hostName = 3;      // Hostname of OM.
+    repeated Port ports = 4;
+}
+
 message Port {
     required string name = 1;
     required uint32 value = 2;

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

@@ -19,21 +19,32 @@ package org.apache.hadoop.hdds.scm.server;
 import com.google.protobuf.BlockingService;
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 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;
 import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
 import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolServerSideTranslatorPB;
 import org.apache.hadoop.hdds.scm.HddsServerUtil;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
+import org.apache.hadoop.hdds.security.x509.SecurityConfig;
+import org.apache.hadoop.hdds.security.x509.certificate.authority.CertificateServer;
+import org.apache.hadoop.hdds.security.x509.certificate.utils.CertificateCodec;
 import org.apache.hadoop.ipc.ProtobufRpcEngine;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.security.KerberosInfo;
+import org.bouncycastle.cert.X509CertificateHolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.hdds.security.x509.certificate.authority.CertificateApprover.ApprovalType.KERBEROS_TRUSTED;
+
 /**
  * The protocol used to perform security related operations with SCM.
  */
@@ -44,15 +55,15 @@ public class SCMSecurityProtocolServer implements SCMSecurityProtocol {
 
   private static final Logger LOGGER = LoggerFactory
       .getLogger(SCMClientProtocolServer.class);
-  private final OzoneConfiguration config;
+  private final SecurityConfig config;
-  private final StorageContainerManager scm;
+  private final CertificateServer certificateServer;
   private final RPC.Server rpcServer;
   private final InetSocketAddress rpcAddress;
 
   SCMSecurityProtocolServer(OzoneConfiguration conf,
-      StorageContainerManager scm) throws IOException {
+      CertificateServer certificateServer) throws IOException {
-    this.config = conf;
+    this.config = new SecurityConfig(conf);
-    this.scm = scm;
+    this.certificateServer = certificateServer;
 
     final int handlerCount =
         conf.getInt(ScmConfigKeys.OZONE_SCM_SECURITY_HANDLER_COUNT_KEY,
@@ -80,7 +91,7 @@ public class SCMSecurityProtocolServer implements SCMSecurityProtocol {
    *
    * @param dnDetails       - DataNode Details.
    * @param certSignReq     - Certificate signing request.
-   * @return byte[]         - SCM signed certificate.
+   * @return String         - SCM signed pem encoded certificate.
    */
   @Override
   public String getDataNodeCertificate(
@@ -88,8 +99,42 @@ public class SCMSecurityProtocolServer implements SCMSecurityProtocol {
       String certSignReq) throws IOException {
     LOGGER.info("Processing CSR for dn {}, UUID: {}", dnDetails.getHostName(),
         dnDetails.getUuid());
-    // TODO: Call scm to sign the csr.
+    Objects.requireNonNull(dnDetails);
-    return null;
+    Future<X509CertificateHolder> future =
+        certificateServer.requestCertificate(certSignReq,
+            KERBEROS_TRUSTED);
+
+    try {
+      return CertificateCodec.getPEMEncodedString(future.get());
+    } catch (InterruptedException | ExecutionException e) {
+      LOGGER.error("getDataNodeCertificate operation failed. ", e);
+      throw new IOException("getDataNodeCertificate operation failed. ", e);
+    }
+  }
+
+  /**
+   * Get SCM signed certificate for OM.
+   *
+   * @param omDetails       - OzoneManager Details.
+   * @param certSignReq     - Certificate signing request.
+   * @return String         - SCM signed pem encoded certificate.
+   */
+  @Override
+  public String getOMCertificate(OzoneManagerDetailsProto omDetails,
+      String certSignReq) throws IOException {
+    LOGGER.info("Processing CSR for om {}, UUID: {}", omDetails.getHostName(),
+        omDetails.getUuid());
+    Objects.requireNonNull(omDetails);
+    Future<X509CertificateHolder> future =
+        certificateServer.requestCertificate(certSignReq,
+            KERBEROS_TRUSTED);
+
+    try {
+      return CertificateCodec.getPEMEncodedString(future.get());
+    } catch (InterruptedException | ExecutionException e) {
+      LOGGER.error("getOMCertificate operation failed. ", e);
+      throw new IOException("getOMCertificate operation failed. ", e);
+    }
   }
 
   public RPC.Server getRpcServer() {

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

@@ -227,15 +227,16 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl
       // TODO: Support Intermediary CAs in future.
       certificateServer.init(new SecurityConfig(conf),
           CertificateServer.CAType.SELF_SIGNED_CA);
+      securityProtocolServer = new SCMSecurityProtocolServer(conf,
+          certificateServer);
     } else {
       // if no Security, we do not create a Certificate Server at all.
       // This allows user to boot SCM without security temporarily
       // and then come back and enable it without any impact.
       certificateServer = null;
+      securityProtocolServer = null;
     }
 
-
-
     eventQueue = new EventQueue();
 
     scmNodeManager = new SCMNodeManager(
@@ -309,11 +310,6 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl
         eventQueue);
     blockProtocolServer = new SCMBlockProtocolServer(conf, this);
     clientProtocolServer = new SCMClientProtocolServer(conf, this);
-    if (OzoneSecurityUtil.isSecurityEnabled(conf)) {
-      securityProtocolServer = new SCMSecurityProtocolServer(conf, this);
-    } else {
-      securityProtocolServer = null;
-    }
     httpServer = new StorageContainerManagerHttpServer(conf);
 
     eventQueue.addHandler(SCMEvents.DATANODE_COMMAND, scmNodeManager);

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMSecurityProtocolServer.java

@@ -40,7 +40,7 @@ public class TestSCMSecurityProtocolServer {
   }
 
   @After
-  public void tearDown() throws Exception {
+  public void tearDown() {
     if (securityProtocolServer != null) {
       securityProtocolServer.stop();
       securityProtocolServer = null;