Use Your Data Flow Delegation Token to Access Oracle Cloud Infrastructure Resources

package example;
 
import com.oracle.bmc.ConfigFileReader;
import com.oracle.bmc.Region;
import com.oracle.bmc.auth.BasicAuthenticationDetailsProvider;
import com.oracle.bmc.auth.ConfigFileAuthenticationDetailsProvider;
import com.oracle.bmc.auth.InstancePrincipalsAuthenticationDetailsProvider;
import com.oracle.bmc.hdfs.BmcProperties;
import com.oracle.bmc.http.ClientConfigurator;
import com.oracle.bmc.http.DefaultConfigurator;
import com.oracle.bmc.http.signing.internal.Constants;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.spark.SparkConf;
 
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.NoSuchElementException;
import javax.annotation.Priority;
import javax.ws.rs.Priorities;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.ClientRequestContext;
import javax.ws.rs.client.ClientRequestFilter;
 
/**
 * Customize the SDK underlying REST client to use the on-behalf-of token when running on
 * Data Flow.
 */
public class OboTokenClientConfigurator implements ClientConfigurator {
 
  // TODO: Set these values for your sepcific OCI environment
  public static final String LOCAL_PROFILE = "DEV"; // TODO <your ~/.oci/config profile>
  private static final String CONFIG_FILE_PATH = ConfigFileReader.DEFAULT_FILE_PATH;
  private static final String CANONICAL_REGION_NAME = Region.US_PHOENIX_1.getRegionId();
 
  private final String delegationTokenPath;
 
  /**
   * Helper function for the Spark Driver to get the token path.
   */
  public static String getDelegationTokenPath() {
    SparkConf conf = new SparkConf();
    try {
      return conf.get("spark.hadoop.fs.oci.client.auth.delegationTokenPath");
    } catch (NoSuchElementException e) {
      return null;
    }
  }
 
  /**
   * Helper function to get the Hadoop configuration for the HDFS BmcFileSystem.
   */
  public static Configuration getConfiguration(Configuration config, String delegationTokenPath) {
    // https://objectstorage.us-phoenix-1.oraclecloud.com
    String domain = "oraclecloud.com";
    String overlayEndpoint = String
        .format("https://objectstorage.%s.%s", CANONICAL_REGION_NAME, domain);
    config.set(BmcProperties.HOST_NAME.getPropertyName(), overlayEndpoint);
    // Data Flow
    if (delegationTokenPath != null) {
      config.set("fs.oci.client.auth.delegationTokenPath", delegationTokenPath);
      config.set(BmcProperties.OBJECT_STORE_CLIENT_CLASS.getPropertyName(),
          "oracle.dfcs.hdfs.DelegationObjectStorageClient");
    } else { // local
      try {
        ConfigFileAuthenticationDetailsProvider provider =
            new ConfigFileAuthenticationDetailsProvider(CONFIG_FILE_PATH, LOCAL_PROFILE);
        config.set(BmcProperties.TENANT_ID.getPropertyName(), provider.getTenantId());
        config.set(BmcProperties.USER_ID.getPropertyName(), provider.getUserId());
        config.set(BmcProperties.FINGERPRINT.getPropertyName(), provider.getFingerprint());
        config.set(BmcProperties.PEM_FILE_PATH.getPropertyName(), provider.getPemFilePath());
      } catch (IOException ex) {
        throw new RuntimeException(ex);
      }
    }
    return config;
  }
 
  /**
   * Helper function to get an environment specific authentication provider.
   */
  public static BasicAuthenticationDetailsProvider getAuthProvider(String delegationTokenPath) {
    if (delegationTokenPath == null) { // local
      try {
        return new ConfigFileAuthenticationDetailsProvider(CONFIG_FILE_PATH, LOCAL_PROFILE);
      } catch (IOException ex) {
        throw new RuntimeException(ex);
      }
    }
    // Data Flow
    return InstancePrincipalsAuthenticationDetailsProvider.builder().build();
  }
 
  /**
   * Helper function to get an environment specific <tt>ClientConfigurator</tt>.
   */
  public static ClientConfigurator getConfigurator(String delegationTokenPath) {
    return (delegationTokenPath == null) ? new DefaultConfigurator() : // local
        new OboTokenClientConfigurator(delegationTokenPath); // Data Flow
  }
 
  /**
   * Helper function to get an environment specific working directory.
   */
  public static String getTempDirectory() {
    if (System.getenv("HOME").equals("/home/dataflow")) {
      return "/opt/spark/work-dir/";
    }
    return System.getProperty("java.io.tmpdir");
  }
 
  public OboTokenClientConfigurator(String delegationTokenPath) {
    this.delegationTokenPath = delegationTokenPath;
  }
 
  @Override
  public void customizeBuilder(ClientBuilder builder) {
  }
 
  @Override
  public void customizeClient(Client client) {
    client.register(new _OboTokenRequestFilter());
  }
 
  @Priority(_OboTokenRequestFilter.PRIORITY)
  class _OboTokenRequestFilter implements ClientRequestFilter {
 
    public static final int PRIORITY = Priorities.AUTHENTICATION - 1;
 
    @Override
    public void filter(final ClientRequestContext requestContext) throws IOException {
      String token = new String(Files.readAllBytes(Paths.get(delegationTokenPath)));
      requestContext.getHeaders().putSingle(Constants.OPC_OBO_TOKEN, token);
    }
  }
}

package example;
 
import com.oracle.bmc.auth.AbstractAuthenticationDetailsProvider;
import com.oracle.bmc.hdfs.BmcFilesystem;
import com.oracle.bmc.http.ClientConfigurator;
import com.oracle.bmc.objectstorage.ObjectStorageClient;
import com.oracle.bmc.objectstorage.requests.GetNamespaceRequest;
import com.oracle.bmc.objectstorage.requests.PutObjectRequest;
import com.oracle.bmc.objectstorage.responses.GetNamespaceResponse;
import com.oracle.bmc.objectstorage.transfer.UploadConfiguration;
import com.oracle.bmc.objectstorage.transfer.UploadManager;
import com.oracle.bmc.objectstorage.transfer.UploadManager.UploadRequest;
 
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkFiles;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
 
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.Arrays;
import java.util.List;
 
// This demo illustrates:
// 1. The delegation token path must be looked up in the Driver and passed to the Executor.
// 2. If the token path is null, assume you are running locally and load an API key.
// 3. Using the HDFS-Connector to create an HDFS FileSystem to connect to Object Storage.
public class Example {
 
  // TODO: Set these values for your sepcific OCI environment
  private static final String NAMESPACE = "<your tenant namespace>";
  private static final String BUCKET_NAME = "output"; // ensure that you create this bucket
 
  private static final String OCI_URI = "oci://" + BUCKET_NAME + "@" + NAMESPACE;
  private static final String SAMPLE_JOB_PATH = "/Example";
  private static final String INPUT_FILE = SAMPLE_JOB_PATH + "/input.dat";
 
  public static void main(String[] args) throws Exception {
    // Get our Spark session.
    SparkConf conf = new SparkConf();
    String master = conf.get("spark.master", "local[*]");
    SparkSession spark = SparkSession.builder().appName("Example").master(master).getOrCreate();
    JavaSparkContext jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());
 
    String delegationTokenPath = OboTokenClientConfigurator.getDelegationTokenPath();
 
    // write a file to Object Storage using an HDFS FileSystem
    try (final BmcFilesystem fs = new BmcFilesystem())
    {
      fs.initialize(new URI(OCI_URI), OboTokenClientConfigurator.getConfiguration(
          jsc.hadoopConfiguration(), delegationTokenPath));
      fs.delete(new Path(SAMPLE_JOB_PATH), true);
      final FSDataOutputStream output = fs.create(new Path(INPUT_FILE));
      output.writeChars("example\npath\ngak\ntest\nexample\ngak\n\ngak");
      output.close();
    }
 
    // Test adding a Spark file, equivalent to spark.files or --files
    jsc.addFile(OCI_URI + INPUT_FILE); // Do not use this for data files
    // Executor -> read a file
    JavaRDD<Integer> rdd = jsc.parallelize(Arrays.asList(1, 2), 2);
    rdd.foreach(item -> executorGetFile());
 
    List<Integer> collection = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
    rdd = jsc.parallelize(collection, 4);
    rdd.foreach(item -> copyFileSample(item, delegationTokenPath));
 
    // Executor -> write a file to Object Storage using an HDFS FileSystem
    rdd = jsc.parallelize(Arrays.asList(1, 2), 2);
    rdd.foreach(item -> fileSystemSample(item, delegationTokenPath));
 
    // ---------------------------------------------------------------------
    // <COPY CODE SNIPPET EXAMPLE #2 HERE>
 
    // ---------------------------------------------------------------------
 
    // ---------------------------------------------------------------------
    // <COPY CODE SNIPPET EXAMPLE #3 HERE>
 
    // ---------------------------------------------------------------------
 
    jsc.stop();
  }
 
  public static void executorGetFile() {
    String filepath = SparkFiles.get("input.dat");
    File file = new File(filepath);
    try {
      String data = IOUtils.toString(new FileInputStream(file), StandardCharsets.UTF_8);
      System.out.println("Read a file:\n" + data);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }
 
  public static void fileSystemSample(int i, String delegationTokenPath) throws Exception {
    // write a file to Object Storage using a FileSystem
    try (final BmcFilesystem fs = new BmcFilesystem())
    {
      fs.initialize(new URI(OCI_URI), OboTokenClientConfigurator.getConfiguration(
          new Configuration(), delegationTokenPath));
      fs.delete(new Path(SAMPLE_JOB_PATH + "/" + i), true);
      final FSDataOutputStream output = fs.create(new Path(INPUT_FILE));
      output.writeChars("example\npath\ngak\ntest\nexample\ngak\n\ngak");
      output.close();
    }
  }
 
  // The driver needs to pass the delegation token path to executors.
  public static void copyFileSample(int i, String delegationTokenPath)
      throws Exception {
    // Create a file to upload.
    String outputFile = MessageFormat.format("{0}.txt", i);
    String outputPath = "/tmp/" + outputFile;
    Writer wr = new FileWriter(outputPath);
    wr.write(String.valueOf(i));
    wr.close();
 
    AbstractAuthenticationDetailsProvider provider =
        OboTokenClientConfigurator.getAuthProvider(delegationTokenPath);
    ClientConfigurator configurator = OboTokenClientConfigurator.getConfigurator(delegationTokenPath);
 
    // Create an object storage client.
    ObjectStorageClient client = ObjectStorageClient.builder()
        .clientConfigurator(configurator)
        .build(provider);
 
    // Look up our namespace.
    GetNamespaceResponse namespaceResponse = client.getNamespace(GetNamespaceRequest.builder().build());
    String namespaceName = namespaceResponse.getValue();
 
    // Upload the file.
    UploadConfiguration uploadConfiguration = UploadConfiguration.builder()
        .allowMultipartUploads(true)
        .allowParallelUploads(true)
        .build();
    UploadManager uploadManager = new UploadManager(client, uploadConfiguration);
    PutObjectRequest request = PutObjectRequest.builder()
        .bucketName(BUCKET_NAME)
        .namespaceName(namespaceName)
        .objectName(outputFile)
        .contentType("text/plain")
        .contentLanguage("EN")
        .contentEncoding("UTF-8")
        .build();
    UploadRequest uploadDetails = UploadRequest.builder(new File(outputPath))
        .allowOverwrite(true)
        .build(request);
    uploadManager.upload(uploadDetails);
  }
}

<!-- Add the Vault SDK dependency to our example project pom.xml below the Secrets SDK dependency -->
    <dependency>
      <groupId>com.oracle.oci.sdk</groupId>
      <artifactId>oci-java-sdk-vault</artifactId>
      <version>${oci-java-sdk-version}</version>
    </dependency>
 
// ---------------------------------------------------------------------
// Add these imports to the top of the file
import com.oracle.bmc.secrets.SecretsClient;
import com.oracle.bmc.secrets.requests.GetSecretBundleRequest;
import com.oracle.bmc.secrets.responses.GetSecretBundleResponse;
import com.oracle.bmc.secrets.model.Base64SecretBundleContentDetails;
import org.apache.commons.codec.binary.Base64;
import com.oracle.bmc.auth.BasicAuthenticationDetailsProvider;
// ---------------------------------------------------------------------
 
// <COPY CODE SNIPPET EXAMPLE #2 HERE>
 
    String passwordOcid = null; // TODO <the vault secret OCID>
 
    BasicAuthenticationDetailsProvider provider =
        OboTokenClientConfigurator.getAuthProvider(delegationTokenPath);
    ClientConfigurator configurator = OboTokenClientConfigurator.getConfigurator(delegationTokenPath);
    SecretsClient secretsClient = SecretsClient.builder().clientConfigurator(configurator).build(provider);
 
    // create get secret bundle request
    GetSecretBundleRequest getSecretBundleRequest = GetSecretBundleRequest
        .builder()
        .secretId(passwordOcid)
        .stage(GetSecretBundleRequest.Stage.Current)
        .build();
 
    // get the secret
    GetSecretBundleResponse getSecretBundleResponse = secretsClient.
        getSecretBundle(getSecretBundleRequest);
 
    // get the bundle content details
    Base64SecretBundleContentDetails base64SecretBundleContentDetails =
        (Base64SecretBundleContentDetails) getSecretBundleResponse.
            getSecretBundle().getSecretBundleContent();
 
    // decode the encoded secret
    byte[] secretValueDecoded = Base64.decodeBase64(base64SecretBundleContentDetails.getContent());
    System.out.println("Secret: " + new String(secretValueDecoded, StandardCharsets.UTF_8));

# Helper Functions
def get_token_path(spark):
    token_key = "spark.hadoop.fs.oci.client.auth.delegationTokenPath"
    token_path = spark.sparkContext.getConf().get(token_key)
    return token_path

def get_authenticated_client(token_path, client):
    import oci
    import os

    if token_path is None:
        # You are running locally, so use our API Key.
        config = oci.config.from_file()
        authenticated_client = client(config)
    else:
        # You are running in Data Flow, so use our Delegation Token.
        with open(token_path) as fd:
            delegation_token = fd.read()
        signer = oci.auth.signers.InstancePrincipalsDelegationTokenSigner(
            delegation_token=delegation_token
        )
        authenticated_client = client(config={}, signer=signer)
    return authenticated_client