flink kerberos认证源码剖析

标签：flink String kerberos System private 源码 null securityConfig

文章目录

01 引言
02 flink的安全机制
03 源码流程分析

3.1 程序入口
3.2 安全模块安装
3.3 模块安装源码

04 文末

01 引言

官方的文档：https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/security/security-kerberos/

我们都知道，如果某个大数据组件（如：hadoop、flink等）需要使用keberos认证，都会使用kinit命令来生成票据，组件一般都会从生成票据的缓存中读取tgt，然后去访问不同的数据源。

但是，一般tgt的有效期一般为24小时，而flink的流式作业都会超过这个时间，tgt很容易就失效了，使用kinit的方式似乎不可行，因此flink也有了自己的方案，可以通过配置kerberos认证信息（如：keytab、krb5、princal等）来实现票据的续签并长期使用，本文来讲讲。

02 flink的安全机制

通过阅读官方的文档，我们知道 Kerberos 用户凭证是组件之间共享的配置，每个组件会显式地使用它，博主理解的就是flink操作hadoop的组件以及不同connector（如：kafka、zookeeper等）的数据源连接，都可以共用同一个kerberos用户凭证，只要该用户分配了所有组件的操作权限即可。

配置示例在flink-conf.yaml看到，如下图：

flink kerberos认证源码剖析_Hadoop

那么flink是如何处理安全认证操作的呢？官网也有描述：

The internal architecture is based on security modules (implementing org.apache.flink.runtime.security.modules.SecurityModule) which are installed at startup.

意思就是，flink 安全内部架构是建立在安全模块上（实现 org.apache.flink.runtime.security.modules.SecurityModule接口），安全模块在 flink 启动过程中被安装。

我们看看SecurityModule这个接口：

flink kerberos认证源码剖析_flink_02

可以看到，实现了SecurityModule有3个模块分别是，HadoopModule、JaasModule、ZookeeperModule。

下面是我整理不同模块的描述：

模块	功能	描述
HadoopModule	该模块使用 Hadoop UserGroupInformation（UGI）类来建立进程范围的登录用户上下文	这还要为了方便与 Hadoop 组件的所有交互，包括 HDFS、HBase 和 YARN
JaasModule	该模块为集群提供动态 jaas 配置，依赖 jass 的组件可直接使用	例如zookeepr和kafka直接依赖即可。flink on yarn模式在容器创建时，会根据kerberos配置自动生成动态的jass配置
ZookeeperModule	该模块配置某些进程范围内 zooKeeper 安全相关的设置	即flink-conf配置文件里面的zookeeper 服务名称（默认为：zookeeper）和 jaas 登录上下文名称（默认为：Client）的关系就在这里体现了

好啦，接下来继续分析这几个模块的初始化的整体流程。

03 源码流程分析

下面是博主整理的源码分析流程图：

flink kerberos认证源码剖析_大数据_03

3.1 程序入口

从上述的流程图可以看到，入口是CliFrontend的main方法：

flink kerberos认证源码剖析_源码分析_04

3.2 安全模块安装

接着进入SecurityUtils的install方法：

flink kerberos认证源码剖析_Hadoop_05

经过代码反查，可以得出程序会读取“security.module.factory.classes”配置，然后自动加载本文一开始说的3个安全模块：

flink kerberos认证源码剖析_源码分析_06

ok，接下来看看各模块的安装逻辑。

3.3 模块安装源码

下面是注释好的安全模块代码描述。

首先是HadoopModule：

/**
 * HadoopModule 是Flink的安全模块之一，用于配置和处理与Hadoop安全性相关的任务。
 *
 * @author : YangLinWei
 * @createTime: 2023/9/7 17:04
 * @version: 1.0.0
 */
public class HadoopModule implements SecurityModule {

    private static final Logger LOG = LoggerFactory.getLogger(HadoopModule.class);

    private final SecurityConfiguration securityConfig; // Flink安全配置

    private final Configuration hadoopConfiguration; // Hadoop配置

    public HadoopModule(
            SecurityConfiguration securityConfiguration, Configuration hadoopConfiguration) {
        this.securityConfig = checkNotNull(securityConfiguration);
        this.hadoopConfiguration = checkNotNull(hadoopConfiguration);
    }

    @VisibleForTesting
    public SecurityConfiguration getSecurityConfig() {
        return securityConfig;
    }

    @Override
    public void install() throws SecurityInstallException {

        // 注入Hadoop的配置信息
        UserGroupInformation.setConfiguration(hadoopConfiguration);

        UserGroupInformation loginUser;

        try {
            // 如果Hadoop启用了kerberos认证，并且配置了Keytab和Principal，那么使用Keytab进行登录
            if (UserGroupInformation.isSecurityEnabled()
                    && !StringUtils.isBlank(securityConfig.getKeytab())
                    && !StringUtils.isBlank(securityConfig.getPrincipal())) {
                String keytabPath = (new File(securityConfig.getKeytab())).getAbsolutePath();

                UserGroupInformation.loginUserFromKeytab(securityConfig.getPrincipal(), keytabPath);

                loginUser = UserGroupInformation.getLoginUser();

                // 补充使用可用的令牌
                String fileLocation =
                        System.getenv(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION);

                if (fileLocation != null) {
                    Credentials credentialsFromTokenStorageFile =
                            Credentials.readTokenStorageFile(
                                    new File(fileLocation), hadoopConfiguration);

                
                    // 因为UGI更喜欢委托令牌，它最终会过期并且不会回退到使用Kerberos票据
                    Credentials credentialsToBeAdded = new Credentials();
                    final Text hdfsDelegationTokenKind = new Text("HDFS_DELEGATION_TOKEN");
                    Collection<Token<? extends TokenIdentifier>> usrTok =
                            credentialsFromTokenStorageFile.getAllTokens();
                    
                    // 如果UGI使用Kerberos Keytab登录，不加载HDFS委托令牌，
                    for (Token<? extends TokenIdentifier> token : usrTok) {
                        if (!token.getKind().equals(hdfsDelegationTokenKind)) {
                            final Text id = new Text(token.getIdentifier());
                            credentialsToBeAdded.addToken(id, token);
                        }
                    }

                    loginUser.addCredentials(credentialsToBeAdded);
                }
            } else {
                // 使用当前用户的凭据登录（例如，票证缓存，操作系统登录）
                // 注意，存储的令牌会自动读取
                try {
                    // 使用反射API获取登录用户对象
                    Method loginUserFromSubjectMethod =
                            UserGroupInformation.class.getMethod(
                                    "loginUserFromSubject", Subject.class);
                    loginUserFromSubjectMethod.invoke(null, (Subject) null);
                } catch (NoSuchMethodException e) {
                    LOG.warn("Could not find method implementations in the shaded jar.", e);
                } catch (InvocationTargetException e) {
                    throw e.getTargetException();
                }

                loginUser = UserGroupInformation.getLoginUser();
            }

            LOG.info("Hadoop user set to {}", loginUser);

            if (HadoopUtils.isKerberosSecurityEnabled(loginUser)) {
                boolean isCredentialsConfigured =
                        HadoopUtils.areKerberosCredentialsValid(
                                loginUser, securityConfig.useTicketCache());

                LOG.info(
                        "Kerberos security is enabled and credentials are {}.",
                        isCredentialsConfigured ? "valid" : "invalid");
            }
        } catch (Throwable ex) {
            throw new SecurityInstallException("Unable to set the Hadoop login user", ex);
        }
    }

    @Override
    public void uninstall() {
        throw new UnsupportedOperationException();
    }
}

接着是JassModule：

/**
 * 负责安装全局的JAAS（Java Authentication and Authorization Service）配置。
 *
 * <p>安装的配置结合了以下因素的登录模块：
 * - 用户提供的JAAS配置文件（如果有）
 * - Kerberos keytab（如果配置了）
 * - 当前环境中的任何缓存的Kerberos凭证
 *
 * <p>该模块还安装了一个默认的JAAS配置文件（如果需要），以确保与ZooKeeper和Kafka的兼容性。注意，JRE实际上会查找多个文件位置。
 * 参考链接:
 * https://docs.oracle.com/javase/7/docs/jre/api/security/jaas/spec/com/sun/security/auth/login/ConfigFile.html
 * https://github.com/apache/kafka/blob/0.9.0/clients/src/main/java/org/apache/kafka/common/security/kerberos/Login.java#L289
 *
 * @author : YangLinWei
 * @createTime: 2023/9/7 17:10
 * @version: 1.0.0
 */
@Internal
public class JaasModule implements SecurityModule {

    private static final Logger LOG = LoggerFactory.getLogger(JaasModule.class);

    static final String JAVA_SECURITY_AUTH_LOGIN_CONFIG = "java.security.auth.login.config";

    static final String JAAS_CONF_RESOURCE_NAME = "flink-jaas.conf";

    private final SecurityConfiguration securityConfig; // Flink安全配置

    private String priorConfigFile;  // 先前的JAAS配置文件
    private javax.security.auth.login.Configuration priorConfig;  // 先前的JAAS配置

    private DynamicConfiguration currentConfig; // 当前的JAAS配置

    /** JAAS文件将安装到的工作目录。 */
    private final String workingDir; // JAAS文件的工作目录

    public JaasModule(SecurityConfiguration securityConfig) {
        this.securityConfig = checkNotNull(securityConfig);
        String[] dirs = splitPaths(securityConfig.getFlinkConfig().getString(CoreOptions.TMP_DIRS));
        /// 至少应该有一个目录。
        checkState(dirs.length > 0);
        this.workingDir = dirs[0];
    }

    @Override
    public void install() {

        // 确保始终定义了配置文件，以与ZooKeeper和Kafka兼容，它们检查系统属性和文件的存在性。
        priorConfigFile = System.getProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, null);
        if (priorConfigFile == null) {
            File configFile = generateDefaultConfigFile(workingDir);
            System.setProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, configFile.getAbsolutePath());
            LOG.info("Jaas file will be created as {}.", configFile);
        }

        // 读取JAAS配置文件
        priorConfig = javax.security.auth.login.Configuration.getConfiguration();

        // 构建动态的JAAS配置
        currentConfig = new DynamicConfiguration(priorConfig);

        // 将配置的JAAS登录上下文与krb5条目关联起来
        AppConfigurationEntry[] krb5Entries = getAppConfigurationEntries(securityConfig);
        if (krb5Entries != null) {
            for (String app : securityConfig.getLoginContextNames()) {
                currentConfig.addAppConfigurationEntry(app, krb5Entries);
            }
        }

        javax.security.auth.login.Configuration.setConfiguration(currentConfig);
    }

    @Override
    public void uninstall() throws SecurityInstallException {
        if (priorConfigFile != null) {
            System.setProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG, priorConfigFile);
        } else {
            System.clearProperty(JAVA_SECURITY_AUTH_LOGIN_CONFIG);
        }
        javax.security.auth.login.Configuration.setConfiguration(priorConfig);
    }

    public DynamicConfiguration getCurrentConfiguration() {
        return currentConfig;
    }

    private static AppConfigurationEntry[] getAppConfigurationEntries(
            SecurityConfiguration securityConfig) {

        AppConfigurationEntry userKerberosAce = null;
        if (securityConfig.useTicketCache()) {
            userKerberosAce = KerberosUtils.ticketCacheEntry();
        }
        AppConfigurationEntry keytabKerberosAce = null;
        if (securityConfig.getKeytab() != null) {
            keytabKerberosAce =
                    KerberosUtils.keytabEntry(
                            securityConfig.getKeytab(), securityConfig.getPrincipal());
        }

        AppConfigurationEntry[] appConfigurationEntry;
        if (userKerberosAce != null && keytabKerberosAce != null) {
            appConfigurationEntry =
                    new AppConfigurationEntry[]{keytabKerberosAce, userKerberosAce};
        } else if (keytabKerberosAce != null) {
            appConfigurationEntry = new AppConfigurationEntry[]{keytabKerberosAce};
        } else if (userKerberosAce != null) {
            appConfigurationEntry = new AppConfigurationEntry[]{userKerberosAce};
        } else {
            return null;
        }

        return appConfigurationEntry;
    }

    /** 生成默认的JAAS配置文件。 */
    private static File generateDefaultConfigFile(String workingDir) {
        checkArgument(workingDir != null, "working directory should not be null.");
        final File jaasConfFile;
        try {
            Path path = Paths.get(workingDir);
            if (Files.notExists(path)) {
                // We intentionally favored Path.toRealPath over Files.readSymbolicLinks as the
                // latter one might return a
                // relative path if the symbolic link refers to it. Path.toRealPath resolves the
                // relative path instead.
                Path parent = path.getParent().toRealPath();
                Path resolvedPath = Paths.get(parent.toString(), path.getFileName().toString());

                path = Files.createDirectories(resolvedPath);
            }
            Path jaasConfPath = Files.createTempFile(path, "jaas-", ".conf");
            try (InputStream resourceStream =
                         JaasModule.class
                                 .getClassLoader()
                                 .getResourceAsStream(JAAS_CONF_RESOURCE_NAME)) {
                Files.copy(resourceStream, jaasConfPath, StandardCopyOption.REPLACE_EXISTING);
            }
            jaasConfFile = new File(workingDir, jaasConfPath.getFileName().toString());
            jaasConfFile.deleteOnExit();
        } catch (IOException e) {
            throw new RuntimeException("unable to generate a JAAS configuration file", e);
        }
        return jaasConfFile;
    }
}

最后是ZookeeperModule：

/**
 * 这个类负责处理ZooKeeper的安全配置，安装全局的ZooKeeper安全配置以支持SASL认证。
 * 它会根据安全配置的设置，设置相应的系统属性，包括是否启用SASL、ZooKeeper服务名称以及登录上下文名称等
 *
 * @author : YangLinWei
 * @createTime: 2023/9/7 17:15
 * @version: 1.0.0
 */
public class ZooKeeperModule implements SecurityModule {

    private static final String ZOOKEEPER_SASL_CLIENT_USERNAME = "zookeeper.sasl.client.username";

    /** 用于设置ZooKeeper是否使用SASL的系统属性。 */
    private static final String ZK_ENABLE_CLIENT_SASL = "zookeeper.sasl.client";

    /** 用于设置期望的ZooKeeper服务名称的系统属性。 */
    private static final String ZK_SASL_CLIENT_USERNAME = "zookeeper.sasl.client.username";

    /** 用于设置要使用的登录上下文名称的系统属性。 */
    private static final String ZK_LOGIN_CONTEXT_NAME = "zookeeper.sasl.clientconfig";

    private final SecurityConfiguration securityConfig; // Flink安全配置

    private String priorSaslEnable; // 先前的ZooKeeper SASL启用状态

    private String priorServiceName; // 先前的ZooKeeper服务名称

    private String priorLoginContextName; // 先前的登录上下文名称

    public ZooKeeperModule(SecurityConfiguration securityConfig) {
        this.securityConfig = checkNotNull(securityConfig);
    }

    @Override
    public void install() throws SecurityInstallException {

        priorSaslEnable = System.getProperty(ZK_ENABLE_CLIENT_SASL, null);
        System.setProperty(
                ZK_ENABLE_CLIENT_SASL, String.valueOf(!securityConfig.isZkSaslDisable()));

        priorServiceName = System.getProperty(ZK_SASL_CLIENT_USERNAME, null);
        if (!"zookeeper".equals(securityConfig.getZooKeeperServiceName())) {
            System.setProperty(ZK_SASL_CLIENT_USERNAME, securityConfig.getZooKeeperServiceName());
        }

        priorLoginContextName = System.getProperty(ZK_LOGIN_CONTEXT_NAME, null);
        if (!"Client".equals(securityConfig.getZooKeeperLoginContextName())) {
            System.setProperty(
                    ZK_LOGIN_CONTEXT_NAME, securityConfig.getZooKeeperLoginContextName());
        }
    }

    @Override
    public void uninstall() throws SecurityInstallException {
        if (priorSaslEnable != null) {
            System.setProperty(ZK_ENABLE_CLIENT_SASL, priorSaslEnable);
        } else {
            System.clearProperty(ZK_ENABLE_CLIENT_SASL);
        }
        if (priorServiceName != null) {
            System.setProperty(ZK_SASL_CLIENT_USERNAME, priorServiceName);
        } else {
            System.clearProperty(ZK_SASL_CLIENT_USERNAME);
        }
        if (priorLoginContextName != null) {
            System.setProperty(ZK_LOGIN_CONTEXT_NAME, priorLoginContextName);
        } else {
            System.clearProperty(ZK_LOGIN_CONTEXT_NAME);
        }
    }
}

04 文末

本文主要讲解了flink kerberos的认证原理以及代码流程分析，希望能帮助到大家，谢谢大家的阅读，本文完！

标签：flink,String,kerberos,System,private,源码,null,securityConfig
From： https://blog.51cto.com/u_15294985/7423617