Addendum for HDDS-11070.

Purify KeyCodec and KeyStorage APIs, clear/add tests for the remaining pieces.
Remove SecurityUtil, replace its functionality with the KeyCodec.
KeyCodec now works with byte[] instead of String.
KeyStorage relies purely on NIO.
Keys related classes moved to hdds-common from hdds-framework.
Intorduced SecurityConstants for string based magic constants in the PEM format.
Changes in SecurityConfig to ensure testability.
Updated APIDocs.

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/SecurityConfig.java

@@ -19,8 +19,10 @@
 
 package org.apache.hadoop.hdds.security;
 
+import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.security.NoSuchAlgorithmException;
 import java.security.Provider;
 import java.security.Security;
 import java.time.Duration;
@@ -32,6 +34,8 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.security.x509.keys.KeyCodec;
+import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslProvider;
 import org.bouncycastle.jce.provider.BouncyCastleProvider;
 import org.slf4j.Logger;
@@ -125,8 +129,8 @@ public class SecurityConfig {
   private final Duration renewalGracePeriod;
   private final boolean isSecurityEnabled;
   private final boolean grpcTlsUseTestCert;
-  private final String externalRootCaPublicKeyPath;
-  private final String externalRootCaPrivateKeyPath;
+  private final Path externalRootCaPublicKeyPath;
+  private final Path externalRootCaPrivateKeyPath;
   private final String externalRootCaCert;
   private final Duration caCheckInterval;
   private final String caRotationTimeOfDay;
@@ -253,12 +257,12 @@ public SecurityConfig(ConfigurationSource configuration) {
     this.externalRootCaCert = configuration.get(
         HDDS_X509_ROOTCA_CERTIFICATE_FILE,
         HDDS_X509_ROOTCA_CERTIFICATE_FILE_DEFAULT);
-    this.externalRootCaPublicKeyPath = configuration.get(
+    this.externalRootCaPublicKeyPath = Paths.get(configuration.get(
         HDDS_X509_ROOTCA_PUBLIC_KEY_FILE,
-        HDDS_X509_ROOTCA_PUBLIC_KEY_FILE_DEFAULT);
-    this.externalRootCaPrivateKeyPath = configuration.get(
+        HDDS_X509_ROOTCA_PUBLIC_KEY_FILE_DEFAULT));
+    this.externalRootCaPrivateKeyPath = Paths.get(configuration.get(
         HDDS_X509_ROOTCA_PRIVATE_KEY_FILE,
-        HDDS_X509_ROOTCA_PRIVATE_KEY_FILE_DEFAULT);
+        HDDS_X509_ROOTCA_PRIVATE_KEY_FILE_DEFAULT));
 
     this.grpcSSLProvider = SslProvider.valueOf(
         configuration.get(HDDS_GRPC_TLS_PROVIDER,
@@ -487,6 +491,14 @@ public String getKeyAlgo() {
     return keyAlgo;
   }
 
+  public KeyCodec keyCodec() throws IOException {
+    try {
+      return new KeyCodec(keyAlgo);
+    } catch (NoSuchAlgorithmException e) {
+      throw new IOException(e);
+    }
+  }
+
   /**
    * Returns the X.509 Signature Algorithm used. This can be changed by setting
    * "hdds.x509.signature.algorithm" to the new name. The default algorithm is
@@ -548,11 +560,16 @@ public SslProvider getGrpcSslProvider() {
     return grpcSSLProvider;
   }
 
-  public String getExternalRootCaPrivateKeyPath() {
+  public boolean useExternalCACertificate(String component) {
+    return component.equals(OzoneConsts.SCM_ROOT_CA_COMPONENT_NAME) &&
+        !externalRootCaCert.isEmpty() && externalRootCaPrivateKeyPath.getNameCount() != 0;
+  }
+
+  public Path getExternalRootCaPrivateKeyPath() {
     return externalRootCaPrivateKeyPath;
   }
 
-  public String getExternalRootCaPublicKeyPath() {
+  public Path getExternalRootCaPublicKeyPath() {
     return externalRootCaPublicKeyPath;
   }
 

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/SecurityConstants.java

@@ -0,0 +1,49 @@
+/*
+ * 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.hadoop.hdds.security;
+
+/**
+ * Class to define constants that are used in relation to different PKIX, PKCS, and CMS Structures as defined by
+ * <a href="https://datatracker.ietf.org/doc/html/rfc7468">RFC-7468</a>.
+ */
+public final class SecurityConstants {
+  private SecurityConstants() { }
+
+  private static final String PEM_PRE_ENCAPSULATION_BOUNDARY_FORMAT = "-----BEGIN %s-----";
+
+  public static final String PEM_POST_ENCAPSULATION_BOUNDARY_FORMAT = "-----END %s-----";
+
+  public static final String PEM_ENCAPSULATION_BOUNDARY_LABEL_PUBLIC_KEY = "PUBLIC KEY";
+
+  public static final String PEM_ENCAPSULATION_BOUNDARY_LABEL_PRIVATE_KEY = "PRIVATE KEY";
+
+  public static final String PEM_PRE_ENCAPSULATION_BOUNDARY_PUBLIC_KEY =
+      String.format(PEM_PRE_ENCAPSULATION_BOUNDARY_FORMAT, PEM_ENCAPSULATION_BOUNDARY_LABEL_PUBLIC_KEY);
+
+  public static final String PEM_POST_ENCAPSULATION_BOUNDARY_PUBLIC_KEY =
+      String.format(PEM_POST_ENCAPSULATION_BOUNDARY_FORMAT, PEM_ENCAPSULATION_BOUNDARY_LABEL_PUBLIC_KEY);
+
+  public static final String PEM_PRE_ENCAPSULATION_BOUNDARY_PRIVATE_KEY =
+      String.format(PEM_PRE_ENCAPSULATION_BOUNDARY_FORMAT, PEM_ENCAPSULATION_BOUNDARY_LABEL_PRIVATE_KEY);
+
+  public static final String PEM_POST_ENCAPSULATION_BOUNDARY_PRIVATE_KEY =
+      String.format(PEM_POST_ENCAPSULATION_BOUNDARY_FORMAT, PEM_ENCAPSULATION_BOUNDARY_LABEL_PRIVATE_KEY);
+
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/keys/KeyCodec.java

@@ -0,0 +1,128 @@
+/*
+ * 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.hadoop.hdds.security.x509.keys;
+
+import org.apache.ratis.util.function.CheckedFunction;
+import org.bouncycastle.util.io.pem.PemObject;
+import org.bouncycastle.util.io.pem.PemReader;
+import org.bouncycastle.util.io.pem.PemWriter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.security.Key;
+import java.security.KeyFactory;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.security.PublicKey;
+import java.security.spec.InvalidKeySpecException;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.security.spec.X509EncodedKeySpec;
+
+// We used UTF-8 before, but a PEM file do contain only readable characters that are in the US_ASCII character set,
+// and UTF-8 is interoperable with US_ASCII in this case.
+// Based on general considerations of RFC-7468 , we stick to the US_ASCII charset for encoding and decoding.
+// See: (https://datatracker.ietf.org/doc/html/rfc7468#section-2)
+import static java.nio.charset.StandardCharsets.US_ASCII;
+import static org.apache.hadoop.hdds.security.SecurityConstants.PEM_ENCAPSULATION_BOUNDARY_LABEL_PRIVATE_KEY;
+import static org.apache.hadoop.hdds.security.SecurityConstants.PEM_ENCAPSULATION_BOUNDARY_LABEL_PUBLIC_KEY;
+
+/**
+ * KeyCodec for encoding and decoding private and public keys.
+ */
+public class KeyCodec {
+  private static final int BUFFER_LEN = 8192;
+
+  private final KeyFactory keyFactory;
+
+  /**
+   * Creates a KeyCodec based on the security configuration.
+   *
+   * @param keyAlgorithm the key algorithm to use.
+   * @throws NoSuchAlgorithmException if the key algorithm specified in the configuration is not available.
+   *
+   * @see <A href="https://docs.oracle.com/en/java/javase/11/docs/specs/security/standard-names.html#keypairgenerator-algorithms">
+   *   Java Security Standard Algorithm Names</A>
+   */
+  public KeyCodec(String keyAlgorithm) throws NoSuchAlgorithmException {
+    keyFactory = KeyFactory.getInstance(keyAlgorithm);
+  }
+
+  /**
+   * Encodes the given public key to PEM format.
+   * @param key the key to encode.
+   * @return the PEM encoded key.
+   * @throws IOException if the encoding fails.
+   */
+  public byte[] encodePublicKey(PublicKey key) throws IOException {
+    return encodeKey(PEM_ENCAPSULATION_BOUNDARY_LABEL_PUBLIC_KEY, key);
+  }
+
+  /**
+   * Encodes the given private key to PEM format.
+   * @param key the key to encode.
+   * @return the PEM encoded key.
+   * @throws IOException if the encoding fails.
+   */
+  public byte[] encodePrivateKey(PrivateKey key) throws IOException {
+    return encodeKey(PEM_ENCAPSULATION_BOUNDARY_LABEL_PRIVATE_KEY, key);
+  }
+
+  /**
+   * Decodes a {@link PrivateKey} from PEM encoded format.
+   * @param encodedKey the PEM encoded key as byte[].
+   * @return a {@link PrivateKey} instance representing the key in the PEM data.
+   * @throws IOException if the decoding fails.
+   */
+  public PrivateKey decodePrivateKey(byte[] encodedKey) throws IOException {
+    return decodeKey(encodedKey, keyFactory::generatePrivate);
+  }
+
+  /**
+   * Decodes a {@link PublicKey} from PEM encoded format.
+   * @param encodedKey the PEM encoded key as byte[].
+   * @return a {@link PublicKey} instance representing the key in the PEM data.
+   * @throws IOException if the decoding fails.
+   */
+  public PublicKey decodePublicKey(byte[] encodedKey) throws IOException {
+    return decodeKey(encodedKey, ks -> keyFactory.generatePublic(new X509EncodedKeySpec(ks.getEncoded())));
+  }
+
+  private byte[] encodeKey(String keyType, Key key) throws IOException {
+    ByteArrayOutputStream bytes = new ByteArrayOutputStream(BUFFER_LEN);
+    try (OutputStreamWriter w = new OutputStreamWriter(bytes, US_ASCII); PemWriter pemWriter = new PemWriter(w)) {
+      pemWriter.writeObject(new PemObject(keyType, key.getEncoded()));
+      pemWriter.flush();
+      return bytes.toByteArray();
+    }
+  }
+
+  private <T> T decodeKey(byte[] encodedKey, CheckedFunction<PKCS8EncodedKeySpec, T, InvalidKeySpecException> generator)
+      throws IOException {
+    try (PemReader pemReader = new PemReader(new InputStreamReader(new ByteArrayInputStream(encodedKey), US_ASCII))) {
+      PemObject keyObject = pemReader.readPemObject();
+      PKCS8EncodedKeySpec pkcs8EncodedKeySpec = new PKCS8EncodedKeySpec(keyObject.getContent());
+      return generator.apply(pkcs8EncodedKeySpec);
+    } catch (InvalidKeySpecException e) {
+      throw new IOException(e);
+    }
+  }
+}