HDDS-594. SCM CA: DN sends CSR and uses certificate issued by SCM. Contributed by Ajay Kumar. (#547)

(cherry picked from commit 064f38b)

Author: ajayydv

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

@@ -19,6 +19,7 @@
 package org.apache.hadoop.hdds;
 
 import javax.management.ObjectName;
+import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.net.InetSocketAddress;
@@ -36,7 +37,15 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+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.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
+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.metrics2.util.MBeans;
 import org.apache.hadoop.net.DNS;
 import org.apache.hadoop.net.NetUtils;
@@ -48,6 +57,8 @@
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_HOST_NAME_KEY;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ENABLED;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ENABLED_DEFAULT;
+
+import org.apache.hadoop.security.UserGroupInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -161,6 +172,29 @@ public static InetSocketAddress getScmAddressForBlockClients(
         .orElse(ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_PORT_DEFAULT));
   }
 
+  /**
+   * Create a scm security client.
+   * @param conf    - Ozone configuration.
+   * @param address - inet socket address of scm.
+   *
+   * @return {@link SCMSecurityProtocol}
+   * @throws IOException
+   */
+  public static SCMSecurityProtocol getScmSecurityClient(
+      OzoneConfiguration conf, InetSocketAddress address) throws IOException {
+    RPC.setProtocolEngine(conf, SCMSecurityProtocolPB.class,
+        ProtobufRpcEngine.class);
+    long scmVersion =
+        RPC.getProtocolVersion(ScmBlockLocationProtocolPB.class);
+    SCMSecurityProtocolClientSideTranslatorPB scmSecurityClient =
+        new SCMSecurityProtocolClientSideTranslatorPB(
+            RPC.getProxy(SCMSecurityProtocolPB.class, scmVersion,
+                address, UserGroupInformation.getCurrentUser(),
+                conf, NetUtils.getDefaultSocketFactory(conf),
+                Client.getRpcTimeout(conf)));
+    return scmSecurityClient;
+  }
+
   /**
    * Retrieve the hostname, trying the supplied config keys in order.
    * Each config value may be absent, or if present in the format

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

@@ -60,17 +60,22 @@ public interface CertificateApprover {
    * @param validFrom - Begin Date
    * @param validTill - End Date
    * @param certificationRequest - Certification Request.
+   * @param scmId - SCM id.
+   * @param clusterId - Cluster id.
    * @return Signed Certificate.
    * @throws IOException - On Error
    * @throws OperatorCreationException - on Error.
    */
+  @SuppressWarnings("ParameterNumber")
   X509CertificateHolder sign(
       SecurityConfig config,
       PrivateKey caPrivate,
       X509CertificateHolder caCertificate,
       Date validFrom,
       Date validTill,
-      PKCS10CertificationRequest certificationRequest)
+      PKCS10CertificationRequest certificationRequest,
+      String scmId,
+      String clusterId)
       throws IOException, OperatorCreationException;
 
 

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

@@ -22,7 +22,10 @@
 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.PKIProfiles.PKIProfile;
+import org.apache.hadoop.hdds.security.x509.keys.SecurityUtil;
 import org.apache.hadoop.util.Time;
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.asn1.x500.style.BCStyle;
 import org.bouncycastle.asn1.x509.AlgorithmIdentifier;
 import org.bouncycastle.asn1.x509.SubjectPublicKeyInfo;
 import org.bouncycastle.cert.X509CertificateHolder;
@@ -67,18 +70,22 @@ public DefaultApprover(PKIProfile pkiProfile, SecurityConfig config) {
    * @param validFrom - Begin Da te
    * @param validTill - End Date
    * @param certificationRequest - Certification Request.
+   * @param scmId - SCM id.
+   * @param clusterId - Cluster id.
    * @return Signed Certificate.
    * @throws IOException - On Error
    * @throws OperatorCreationException - on Error.
    */
+  @SuppressWarnings("ParameterNumber")
   public  X509CertificateHolder sign(
       SecurityConfig config,
       PrivateKey caPrivate,
       X509CertificateHolder caCertificate,
       Date validFrom,
       Date validTill,
-      PKCS10CertificationRequest certificationRequest)
-      throws IOException, OperatorCreationException {
+      PKCS10CertificationRequest certificationRequest,
+      String scmId,
+      String clusterId) throws IOException, OperatorCreationException {
 
     AlgorithmIdentifier sigAlgId = new
         DefaultSignatureAlgorithmIdentifierFinder().find(
@@ -91,6 +98,29 @@ public  X509CertificateHolder sign(
     SubjectPublicKeyInfo keyInfo =
         certificationRequest.getSubjectPublicKeyInfo();
 
+    // Get scmId and cluster Id from subject name.
+    X500Name x500Name = certificationRequest.getSubject();
+    String csrScmId = x500Name.getRDNs(BCStyle.OU)[0].getFirst().getValue().
+        toASN1Primitive().toString();
+    String csrClusterId = x500Name.getRDNs(BCStyle.O)[0].getFirst().getValue().
+        toASN1Primitive().toString();
+
+    if (!scmId.equals(csrScmId) || !clusterId.equals(csrClusterId)) {
+      if (csrScmId.equalsIgnoreCase("null") &&
+          csrClusterId.equalsIgnoreCase("null")) {
+        // Special case to handle DN certificate generation as DN might not know
+        // scmId and clusterId before registration. In secure mode registration
+        // will succeed only after datanode has a valid certificate.
+        String cn = x500Name.getRDNs(BCStyle.CN)[0].getFirst().getValue()
+            .toASN1Primitive().toString();
+        x500Name = SecurityUtil.getDistinguishedName(cn, scmId, clusterId);
+      } else {
+        // Throw exception if scmId and clusterId doesn't match.
+        throw new SCMSecurityException("ScmId and ClusterId in CSR subject" +
+            " are incorrect.");
+      }
+    }
+
     RSAKeyParameters rsa =
         (RSAKeyParameters) PublicKeyFactory.createKey(keyInfo);
     if (rsa.getModulus().bitLength() < config.getSize()) {
@@ -104,7 +134,7 @@ public  X509CertificateHolder sign(
             BigInteger.valueOf(Time.monotonicNowNanos()),
             validFrom,
             validTill,
-            certificationRequest.getSubject(), keyInfo);
+            x500Name, keyInfo);
 
     ContentSigner sigGen = new BcRSAContentSignerBuilder(sigAlgId, digAlgId)
         .build(asymmetricKP);

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

@@ -227,7 +227,7 @@ public Future<X509CertificateHolder> requestCertificate(
         X509CertificateHolder xcert = approver.sign(config,
             getCAKeys().getPrivate(),
             getCACertificate(), java.sql.Date.valueOf(beginDate),
-            java.sql.Date.valueOf(endDate), csr);
+            java.sql.Date.valueOf(endDate), csr, scmID, clusterID);
         store.storeValidCertificate(xcert.getSerialNumber(),
             CertificateCodec.getX509Certificate(xcert));
         xcertHolder.complete(xcert);

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/certificates/utils/CertificateSignRequest.java

@@ -269,10 +269,6 @@ public PKCS10CertificationRequest build() throws SCMSecurityException {
       Preconditions.checkNotNull(key, "KeyPair cannot be null");
       Preconditions.checkArgument(Strings.isNotBlank(subject), "Subject " +
           "cannot be blank");
-      Preconditions.checkArgument(Strings.isNotBlank(clusterID), "Cluster ID " +
-          "cannot be blank");
-      Preconditions.checkArgument(Strings.isNotBlank(scmID), "SCM ID cannot " +
-          "be blank");
 
       try {
         CertificateSignRequest csr = new CertificateSignRequest(subject, scmID,

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/security/x509/certificate/authority/MockApprover.java

@@ -47,9 +47,10 @@ public MockApprover(PKIProfile pkiProfile, SecurityConfig config) {
 
   @Override
   public X509CertificateHolder sign(SecurityConfig config, PrivateKey caPrivate,
-                                    X509CertificateHolder caCertificate,
-                                    Date validFrom, Date validTill,
-                                    PKCS10CertificationRequest request)
+      X509CertificateHolder caCertificate,
+      Date validFrom, Date validTill,
+      PKCS10CertificationRequest request,
+      String scmId, String clusterId)
       throws IOException, OperatorCreationException {
     return null;
   }

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/security/x509/certificate/authority/TestDefaultCAServer.java

@@ -25,6 +25,7 @@
 import org.apache.hadoop.hdds.security.x509.SecurityConfig;
 import org.apache.hadoop.hdds.security.x509.certificates.utils.CertificateSignRequest;
 import org.apache.hadoop.hdds.security.x509.keys.HDDSKeyGenerator;
+import org.apache.hadoop.test.LambdaTestUtils;
 import org.bouncycastle.cert.X509CertificateHolder;
 import org.bouncycastle.pkcs.PKCS10CertificationRequest;
 import org.junit.Before;
@@ -139,14 +140,57 @@ public void testMissingKey() {
   public void testRequestCertificate() throws IOException,
       ExecutionException, InterruptedException,
       NoSuchProviderException, NoSuchAlgorithmException {
+    String scmId =  RandomStringUtils.randomAlphabetic(4);
+    String clusterId =  RandomStringUtils.randomAlphabetic(4);
+    KeyPair keyPair =
+        new HDDSKeyGenerator(conf).generateKey();
+    PKCS10CertificationRequest csr = new CertificateSignRequest.Builder()
+        .addDnsName("hadoop.apache.org")
+        .addIpAddress("8.8.8.8")
+        .setCA(false)
+        .setClusterID(clusterId)
+        .setScmID(scmId)
+        .setSubject("Ozone Cluster")
+        .setConfiguration(conf)
+        .setKey(keyPair)
+        .build();
+
+    // Let us convert this to a string to mimic the common use case.
+    String csrString = CertificateSignRequest.getEncodedString(csr);
+
+    CertificateServer testCA = new DefaultCAServer("testCA",
+        clusterId, scmId, caStore);
+    testCA.init(new SecurityConfig(conf),
+        CertificateServer.CAType.SELF_SIGNED_CA);
+
+    Future<X509CertificateHolder> holder = testCA.requestCertificate(csrString,
+        CertificateApprover.ApprovalType.TESTING_AUTOMATIC);
+    // Right now our calls are synchronous. Eventually this will have to wait.
+    assertTrue(holder.isDone());
+    assertNotNull(holder.get());
+  }
+
+  /**
+   * Tests that we are able
+   * to create a Test CA, creates it own self-Signed CA and then issue a
+   * certificate based on a CSR when scmId and clusterId are not set in
+   * csr subject.
+   * @throws SCMSecurityException - on ERROR.
+   * @throws ExecutionException - on ERROR.
+   * @throws InterruptedException - on ERROR.
+   * @throws NoSuchProviderException - on ERROR.
+   * @throws NoSuchAlgorithmException - on ERROR.
+   */
+  @Test
+  public void testRequestCertificateWithInvalidSubject() throws IOException,
+      ExecutionException, InterruptedException,
+      NoSuchProviderException, NoSuchAlgorithmException {
     KeyPair keyPair =
         new HDDSKeyGenerator(conf).generateKey();
     PKCS10CertificationRequest csr = new CertificateSignRequest.Builder()
         .addDnsName("hadoop.apache.org")
         .addIpAddress("8.8.8.8")
         .setCA(false)
-        .setClusterID("ClusterID")
-        .setScmID("SCMID")
         .setSubject("Ozone Cluster")
         .setConfiguration(conf)
         .setKey(keyPair)
@@ -168,4 +212,40 @@ public void testRequestCertificate() throws IOException,
     assertNotNull(holder.get());
   }
 
+  @Test
+  public void testRequestCertificateWithInvalidSubjectFailure()
+      throws Exception {
+    KeyPair keyPair =
+        new HDDSKeyGenerator(conf).generateKey();
+    PKCS10CertificationRequest csr = new CertificateSignRequest.Builder()
+        .addDnsName("hadoop.apache.org")
+        .addIpAddress("8.8.8.8")
+        .setCA(false)
+        .setScmID("wrong one")
+        .setClusterID("223432rf")
+        .setSubject("Ozone Cluster")
+        .setConfiguration(conf)
+        .setKey(keyPair)
+        .build();
+
+    // Let us convert this to a string to mimic the common use case.
+    String csrString = CertificateSignRequest.getEncodedString(csr);
+
+    CertificateServer testCA = new DefaultCAServer("testCA",
+        RandomStringUtils.randomAlphabetic(4),
+        RandomStringUtils.randomAlphabetic(4), caStore);
+    testCA.init(new SecurityConfig(conf),
+        CertificateServer.CAType.SELF_SIGNED_CA);
+
+    LambdaTestUtils.intercept(ExecutionException.class, "ScmId and " +
+            "ClusterId in CSR subject are incorrect",
+        () -> {
+          Future<X509CertificateHolder> holder =
+              testCA.requestCertificate(csrString,
+                  CertificateApprover.ApprovalType.TESTING_AUTOMATIC);
+          holder.isDone();
+          holder.get();
+        });
+  }
+
 }

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/security/x509/certificates/TestCertificateSignRequest.java

@@ -213,24 +213,6 @@ public void testGenerateCSRWithInvalidParams() throws NoSuchProviderException,
       builder.setSubject(subject);
     }
 
-    // Now try with blank/null SCM ID
-    try {
-      builder.setScmID(null);
-      builder.build();
-      Assert.fail("Null/Blank SCM ID should have thrown.");
-    } catch (IllegalArgumentException e) {
-      builder.setScmID(scmID);
-    }
-
-    // Now try with blank/null SCM ID
-    try {
-      builder.setClusterID(null);
-      builder.build();
-      Assert.fail("Null/Blank Cluster ID should have thrown.");
-    } catch (IllegalArgumentException e) {
-      builder.setClusterID(clusterID);
-    }
-
     // Now try with invalid IP address
     try {
       builder.addIpAddress("255.255.255.*");