# 获取Flink作业委托临时凭证用于访问其他云服务
#### 什么是临时凭证？
临时安全凭证是具备临时访问权限的身份凭证，包括临时AK/SK和SecurityToken，临时AK/SK和SecurityToken必须同时使用。
了解[临时安全凭证](https://support.huaweicloud.com/iam_faq/iam_01_0620.html)。
#### 功能描述
AI DataLake提供了一个通用接口，可用于获取作业所使用委托的临时凭证。作业使用的委托在创建端点时配置，详见[约束限制]。该接口将获取到的作业委托的临时凭证封装到com.huaweicloud.sdk.core.auth.BasicCredentials类中。
- 获取到的委托的临时认证封装到com.huaweicloud.sdk.core.auth.ICredentialProvider接口的getCredentials()返回值中。
- 返回类型为com.huaweicloud.sdk.core.auth.BasicCredentials。
- 仅支持获取AK、SK、SecurityToken。
- ICredentialProvider的实现已内置于AI DataLake Flink引擎中，并通过SPI机制注册，用户作业中无需自行实现该接口，直接使用"示例代码"章节提供的工具类加载获取即可。
- 获取到AK、SK、SecurityToken后，请参考[如何使用凭据管理服务替换硬编码的数据库账号密码](https://support.huaweicloud.com/bestpractice-dew/dew_06_0014.html)查询凭据。
  使用临时AK、SK访问华为云凭据管理服务（CSMS），需在参考文档中getCsmsClient()方法的配置基础上，添加临时凭证必需的SecurityToken配置：
  ```
  private static CsmsClient getCsmsClient() {
          BasicCredentials auth = new BasicCredentials()
                  .withAk(ACCESS_KEY)
                  .withSk(SECRET_ACCESS_KEY)
                  .withSecurityToken(securityToken)
                  .withProjectId(PROJECT_ID);
  ```
  
 
 #### 约束限制
- 仅支持Flink1.20版本使用委托授权访问临时凭证。AI DataLake Flink引擎当前为Flink1.20版本，无需额外选择作业引擎版本。
- 已在创建端点时配置允许AI DataLake访问DEW的委托信息（端点凭证配置中的委托名称）。作业获取临时凭证时使用该委托：
  - 提交作业时未指定作业级委托，则默认使用端点配置的委托。
  
  - 提交作业时，也可以通过作业请求参数job_agency指定作业级委托。指定的委托必须已存在，且不能使用系统管理委托。
  
  - 委托的临时凭证由平台自动注入作业运行环境，作业代码中通过工具类接口获取即可，无需在作业配置中额外指定委托参数。 委托权限示例请参考[创建自定义委托](https://support.huaweicloud.com/usermanual-aidatalake/aidatalake_01_0078.html)和[常见场景的委托权限策略](https://support.huaweicloud.com/usermanual-aidatalake/aidatalake_01_0083.html)。
    
   
- AI DataLake Flink引擎基础镜像内已内置3.1.177版本的huaweicloud-sdk-core（版本随引擎发布变化，以引擎实际内置版本为准），无需重复安装。请在pom文件配置中将该依赖声明为provided，避免打包进用户Jar。
 
#### 准备环境
已安装和配置IntelliJ IDEA等开发工具以及安装JDK和Maven。
Maven工程的pom.xml文件配置请参考[JAVA样例代码](https://support.huaweicloud.com/devg-dli/dli_09_0191.html#dli_09_0191__section865362419491)中"pom文件配置"说明。
pom文件配置中依赖包
```
<dependency>
            <groupId>com.huaweicloud.sdk</groupId>
            <artifactId>huaweicloud-sdk-core</artifactId>
            <version>3.1.177</version>
            <scope>provided</scope>
        </dependency>
```
如需使用临时凭证访问华为云凭据管理服务（DEW CSMS）查询凭据，还需增加huaweicloud-sdk-csms依赖，并排除其传递依赖的huaweicloud-sdk-core，保持SDK core由引擎提供，避免用户Jar中出现两份SDK core：
```
<dependency>
            <groupId>com.huaweicloud.sdk</groupId>
            <artifactId>huaweicloud-sdk-csms</artifactId>
            <version>3.1.177</version>
            <exclusions>
                <exclusion>
                    <groupId>com.huaweicloud.sdk</groupId>
                    <artifactId>huaweicloud-sdk-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
```
#### 示例代码
本章节JAVA样例代码演示如何获取BasicCredentials，以及获取作业委托的AK、SK、SecurityToken。
示例代码中的CSMS_ENDPOINT、PROJECT_ID、SECRET_NAME需按实际情况替换：
- **CSMS_ENDPOINT** ：华为云CSMS服务访问终端地址，与目标凭据所在区域对应，格式为https://kms.\<region\>.myhuaweicloud.com，可从[DEW API参考"使用前必读"的"终端节点"章节](https://support.huaweicloud.com/api-dew/dew_02_0052.html)查询。
- **PROJECT_ID**：华为云局点项目ID，取CSMS_ENDPOINT对应区域的项目ID，在控制台右上角"用户名 \> 我的凭证 \> 项目列表"中查看。
- **SECRET_NAME**：DEW凭据管理服务中目标凭据的名称。
- **Flink UDF 获取作业委托凭证**
  ```
  package com.aidatalake.flink.demo;
  import static com.aidatalake.flink.demo.utils.AIDataLakeJobAgencyCredentialUtils.getICredentialProvider;
  import com.huaweicloud.sdk.core.auth.BasicCredentials;
  import com.huaweicloud.sdk.core.auth.ICredentialProvider;
  import com.huaweicloud.sdk.csms.v1.CsmsClient;
  import com.huaweicloud.sdk.csms.v1.model.Secret;
  import com.huaweicloud.sdk.csms.v1.model.ShowSecretRequest;
  import com.huaweicloud.sdk.csms.v1.model.ShowSecretResponse;
  import org.apache.flink.table.functions.FunctionContext;
  import org.apache.flink.table.functions.ScalarFunction;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import java.util.List;
  public class GetUserAgencyCredentialUDF extends ScalarFunction {
      private static final Logger LOG = LoggerFactory.getLogger(GetUserAgencyCredentialUDF.class);
      // 按实际情况替换：DEW CSMS接入地址与项目ID
      private static final String CSMS_ENDPOINT = "https://kms.<region>.myhuaweicloud.com";
      private static final String PROJECT_ID = "your-project-id";
      ICredentialProvider credentialProvider;
      CsmsClient csmsClient;
      @Override
      public void open(FunctionContext context) throws Exception {
          // 通过SPI获取作业委托临时凭证，无需在作业配置中额外指定委托参数
          credentialProvider = getICredentialProvider();
          // 临时AK/SK和SecurityToken已封装在BasicCredentials中（SecurityToken为临时凭证必需项），指定项目ID后构建CSMS客户端
          BasicCredentials basicCredentials = (BasicCredentials) credentialProvider.getCredentials();
          basicCredentials = basicCredentials.withProjectId(PROJECT_ID);
          csmsClient = CsmsClient.newBuilder()
                  .withCredential(basicCredentials)
                  .withEndpoints(List.of(CSMS_ENDPOINT))
                  .build();
      }
      public String eval(String secretName) {
          ShowSecretResponse response = csmsClient.showSecret(
                  new ShowSecretRequest().withSecretName(secretName));
          Secret secret = response.getSecret();
          if (secret == null) {
              throw new RuntimeException("DEW secret not found: " + secretName);
          }
          LOG.info("Accessed DEW CSMS with agency temp credentials, secret name={}", secret.getName());
          // 返回凭据元数据证明使用作业委托的临时凭证访问DEW成功；实际业务中可在此获取凭据值用于后续访问其他云服务
          return secret.getName();
      }
  }
  ```
  将上述UDF类及工具类打包为Jar并上传至OBS后，即可在Flink SQL作业中通过CREATE TEMPORARY FUNCTION语句注册该UDF，并在SELECT中使用作业委托的临时凭证查询DEW凭据：
  ```
  CREATE TEMPORARY FUNCTION get_dew_credential AS 'com.aidatalake.flink.demo.GetUserAgencyCredentialUDF'
    USING JAR 'obs://<bucket>/<path>/agency-credential-udf.jar';
  SELECT get_dew_credential('secret-name');
  ```
  USING JAR指定的Jar在SQL编译时自动下载注册，无需在作业配置中额外添加依赖；作业运行成功即表示已使用作业委托的临时凭证成功访问DEW。
  

- **Flink Jar作业获取作业委托凭证**
  ```
  package com.aidatalake.flink.demo;
  import static com.aidatalake.flink.demo.utils.AIDataLakeJobAgencyCredentialUtils.getICredentialProvider;
  import com.huaweicloud.sdk.core.auth.BasicCredentials;
  import com.huaweicloud.sdk.core.auth.ICredentialProvider;
  import com.huaweicloud.sdk.csms.v1.CsmsClient;
  import com.huaweicloud.sdk.csms.v1.model.Secret;
  import com.huaweicloud.sdk.csms.v1.model.ShowSecretRequest;
  import com.huaweicloud.sdk.csms.v1.model.ShowSecretResponse;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import java.util.List;
  public class GetUserCredentialsFlinkStream {
      // 按实际情况替换：DEW CSMS接入地址、项目ID与凭据名
      private static final String CSMS_ENDPOINT = "https://kms.<region>.myhuaweicloud.com";
      private static final String PROJECT_ID = "your-project-id";
      private static final String SECRET_NAME = "your-secret-name";
      public static void main(String[] args) throws Exception {
          StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
          streamEnv.fromElements(1)
                  .map(i -> {
                      ICredentialProvider credentialProvider = getICredentialProvider();
                      // 临时AK/SK和SecurityToken已封装在BasicCredentials中（SecurityToken为临时凭证必需项），指定项目ID后构建CSMS客户端
                      BasicCredentials basicCredentials = (BasicCredentials) credentialProvider.getCredentials();
                      basicCredentials = basicCredentials.withProjectId(PROJECT_ID);
                      CsmsClient client = CsmsClient.newBuilder()
                              .withCredential(basicCredentials)
                              .withEndpoints(List.of(CSMS_ENDPOINT))
                              .build();
                      // 使用作业委托的临时凭证查询DEW凭据
                      ShowSecretResponse response = client.showSecret(
                              new ShowSecretRequest().withSecretName(SECRET_NAME));
                      Secret secret = response.getSecret();
                      if (secret == null) {
                          throw new RuntimeException("DEW secret not found: " + SECRET_NAME);
                      }
                      return "Accessed DEW CSMS with job agency credential, secret name=" + secret.getName()
                              + ", id=" + secret.getId();
                  })
                  .print();
          streamEnv.execute("GetUserCredentialsFlinkStream");
      }
  }
  ```
  

- **获取作业委托的工具类**
  ```
  package com.aidatalake.flink.demo.utils;
  import com.huaweicloud.sdk.core.auth.ICredentialProvider;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.ServiceLoader;
  public class AIDataLakeJobAgencyCredentialUtils {
      public static ICredentialProvider getICredentialProvider() {
          List<ICredentialProvider> credentialProviders = new ArrayList<>();
          ServiceLoader.load(ICredentialProvider.class, StreamExecutionEnvironment.class.getClassLoader())
              .iterator()
              .forEachRemaining(credentialProviders::add);
          if (credentialProviders.size() != 1) {
              throw new RuntimeException("Failed to obtain temporary user credential");
          }
          return credentialProviders.get(0);
      }
  }
  ```