Apache Kylin CVE-2022-24697/CVE-2022-43396 漏洞分析与CodeQL 验证

在官方漏洞报告中，说CVE-2022-43396是CVE-2022-24697的绕过。实际情况却是：两个漏洞均是发送在 Kylin 的 cube build 功能中，但CVE-2022-24697是参数可控；CVE-2022-43396是命令可控。

下文以CVE-2022-24697 为例，主要目的是跟踪下数据流，看看能否用CodeQL 复现。

1. 漏洞概况

影响版本：
Kylin 2.x & Kylin 3.x & 4.x should upgrade to 4.0.2

2. 漏洞分析

2.1 漏洞环境搭建

同前文，不赘述

参考[1] 中的步骤，在Cube Designer Configuration Overwrites 中配置如下：

1
2
kylin.engine.spark-conf.spark.driver.memory
512M' `touch /tmp/hacked` --name 'a

upload successful

upload successful

2.2 漏洞原理分析

整个逻辑涉及到几个线程

  1. web api CubeController#updateCubeDesc 更新CubeDesc 配置信息，在其config.overrides，config是个KylinConfigExt 类型；随后异步更新CubeInstance
  2. web api CubeController#rebuild 生成一条rebuild Job，写入ExecutableManager 的队列中
  3. DefaultScheduler 循环线程，始终从ExecutableManager 中获取job 信息，根据job 中的cubename从CubeManager 获取Cube 信息，获取步骤1 中的config.overrides 进行命令拼接，导致命令执行

2.2.1 CubeController#updateCubeDesc

CubeController#updateCubeDesc

1
2
3
4
5
6
@RequestMapping(value = "", method = { RequestMethod.PUT }, produces = { "application/json" })
@ResponseBody
public CubeRequest updateCubeDesc(@RequestBody CubeRequest cubeRequest) throws JsonProcessingException {
    CubeDesc desc = deserializeCubeDesc(cubeRequest);
        CubeInstance cube = cubeService.getCubeManager().getCube(cubeRequest.getCubeName());
        desc = cubeService.updateCubeAndDesc(cube, desc, projectName, true);

CubeService.java

1
2
3
4
public class CubeService extends BasicService implements InitializingBean {
    public CubeDesc updateCubeAndDesc(CubeInstance cube, CubeDesc desc, String newProjectName, boolean forceUpdate)
        CubeDesc updatedCubeDesc = getCubeDescManager().updateCubeDesc(desc);

CubeDescManager.java

1
2
3
public class CubeDescManager {
    public CubeDesc updateCubeDesc(CubeDesc desc) throws IOException {
            crud.save(desc);
  • 更新CubeDesc
CubeDesc to CubeInstance

但是从下文中rebuild 中得知，是直接获取的cubeInstance，那么是CubeDesc 是如何修改CubeInstance 的呢？

upload successful

upload successful

upload successful

  1. 更新完cubeDesc后，会触发EntityChange 事件
  2. 最终调用cubeManager.reloadCubeQuietly 重新加载cube

这中间涉及到异步，CodeQL 基本无望了

2.2.2 CubeController#rebuild

upload successful

CubeController#rebuild

1
2
3
4
5
6
7
8
9
10
11
12
13
14
   @RequestMapping(value = "/{cubeName}/rebuild", method = { RequestMethod.PUT }, produces = { "application/json" })
   @ResponseBody
   public JobInstance rebuild(@PathVariable String cubeName, @RequestBody JobBuildRequest req) {
       return buildInternal(cubeName, new TSRange(req.getStartTime(), req.getEndTime()), null, null, null,
               req.getBuildType(), req.isForce() || req.isForceMergeEmptySegment(), req.getPriorityOffset());

   private JobInstance buildInternal(String cubeName, TSRange tsRange, SegmentRange segRange, //
           Map<Integer, Long> sourcePartitionOffsetStart, Map<Integer, Long> sourcePartitionOffsetEnd,
           String buildType, boolean force, Integer priorityOffset) {
       try {
           CubeInstance cube = jobService.getCubeManager().getCube(cubeName);
           return jobService.submitJob(cube, tsRange, segRange, sourcePartitionOffsetStart, sourcePartitionOffsetEnd,
		...
}
  • 通过jobService.getCubeManager().getCube(cubeName) 获取cube

JobService#submitJob

1
2
3
4
5
6
7
8
9
10
11
public JobInstance submitJob(CubeInstance cube, TSRange tsRange, SegmentRange segRange, //
    JobInstance jobInstance = submitJobInternal(cube, tsRange, segRange, sourcePartitionOffsetStart,

public JobInstance submitJobInternal(CubeInstance cube, TSRange tsRange, SegmentRange segRange, //
        Map<Integer, Long> sourcePartitionOffsetStart, Map<Integer, Long> sourcePartitionOffsetEnd, //
        CubeBuildTypeEnum buildType, boolean force, String submitter, Integer priorityOffset) throws IOException {
        ...
            newSeg = getCubeManager().appendSegment(cube, src);
            job = EngineFactory.createBatchCubingJob(newSeg, submitter, priorityOffset);
        ...
        getExecutableManager().addJob(job);

ExecutableManager.java

1
2
3
4
5
6
7
8
9
10
11
12
13
public class ExecutableManager {
    private final KylinConfig config;
    private final ExecutableDao executableDao;
    
    public void addJob(AbstractExecutable executable) {
        try {
            executable.initConfig(config);
            addJobOutput(executable);
            executableDao.addJob(parse(executable));

    public List<String> getAllJobIdsInCache() {
        return executableDao.getJobIdsInCache();
    }

ExecutableDao.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class ExecutableDao {
    private CaseInsensitiveStringCache<ExecutablePO> executableDigestMap;
    
    private ExecutableDao(KylinConfig config) throws IOException {
        this.executableDigestMap = new CaseInsensitiveStringCache<>(config, "execute");

    public List<String> getJobIdsInCache() {
        Set<String> idSet = executableDigestMap.keySet();
        return Lists.newArrayList(idSet);
    }
    
    public ExecutablePO addJob(ExecutablePO job) throws PersistentException {
    	...
            executableDigestMap.put(job.getId(), job);
        ...
    }
  • ExecutableDao 存在个map，保存所有的Executeable 任务
  • addJob 新增job任务接口
  • getJobIdsInCache 获取所有任务接口

2.2.2 DefaultScheduler 定时任务线程

upload successful

1
2
3
4
5
6
7
8
9
10
execute:90, CliCommandExecutor (org.apache.kylin.common.util)
runSparkSubmit:282, NSparkExecutable (org.apache.kylin.engine.spark.job)
doWork:168, NSparkExecutable (org.apache.kylin.engine.spark.job)
execute:206, AbstractExecutable (org.apache.kylin.job.execution)
doWork:94, DefaultChainedExecutable (org.apache.kylin.job.execution)
execute:206, AbstractExecutable (org.apache.kylin.job.execution)
run:113, DefaultScheduler$JobRunner (org.apache.kylin.job.impl.threadpool)
runWorker:1149, ThreadPoolExecutor (java.util.concurrent)
run:624, ThreadPoolExecutor$Worker (java.util.concurrent)
run:748, Thread (java.lang)

中间变量为 CubeInstance.config.overrides 内

NSparkExecutable.java

1
2
3
4
5
6
7
protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
    //context.setLogPath(getSparkDriverLogHdfsPath(context.getConfig()));
    CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
    CubeInstance cube = cubeMgr.getCube(this.getCubeName());
    KylinConfig config = cube.getConfig();
    this.setLogPath(getSparkDriverLogHdfsPath(context.getConfig()));
    config = wrapConfig(config);
ThreadPool 任务从何而来

DefaultScheduler.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class DefaultScheduler implements Scheduler<AbstractExecutable> {
    private static DefaultScheduler INSTANCE;

    public synchronized void init(JobEngineConfig jobEngineConfig, JobLock lock) throws SchedulerException {
    	...
        jobPool = new ThreadPoolExecutor(corePoolSize, corePoolSize, Long.MAX_VALUE, TimeUnit.DAYS,
                new SynchronousQueue<Runnable>());
        ...
        JobExecutor jobExecutor = new JobExecutor() {
            @Override
            public void execute(AbstractExecutable executable) {
                jobPool.execute(new JobRunner(executable));
            }
        };
        ...
        fetcher = jobEngineConfig.getJobPriorityConsidered()
                ? new PriorityFetcherRunner(jobEngineConfig, context, jobExecutor)
                : new DefaultFetcherRunner(jobEngineConfig, context, jobExecutor);
  • jobPool.execute 是从pool 里面添加任务的接口
  • 得看后续jobExecutor 何处调用execute

FetcherRunner.java

1
2
3
4
5
6
7
8
9
10
11
12
public abstract class FetcherRunner implements Runnable {
	...
    public FetcherRunner(JobEngineConfig jobEngineConfig, DefaultContext context, JobExecutor jobExecutor) {
        this.jobEngineConfig = jobEngineConfig;
        this.context = context;
        this.jobExecutor = jobExecutor;
    }
    ...
    protected void addToJobPool(AbstractExecutable executable, int priority) {
    	...
            jobExecutor.execute(executable);

1
2
3
4
5
public class DefaultFetcherRunner extends FetcherRunner {
    synchronized public void run() {
            for (final String id : getExecutableManager().getAllJobIdsInCache()) {
                final AbstractExecutable executable = getExecutableManager().getJob(id);
                addToJobPool(executable, executable.getDefaultPriority());
  • DefaultFetcherRunner 是个Runnable，其run 逻辑为循环从getExecutableManager() 获取任务，并且添加到jobExecutor 中去执行
1
2
3
4
5
6
7
8
9
10
11
private class JobRunner implements Runnable {
	private final AbstractExecutable executable;

    public JobRunner(AbstractExecutable executable) {
        this.executable = executable;
    }
    
    public void run() {
        try (SetThreadName ignored = new SetThreadName("Scheduler %s Job %s",
                System.identityHashCode(DefaultScheduler.this), executable.getId())) {
            executable.execute(context);
  • 所以最终调用的是executable#execute 接口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@EnableAspectJAutoProxy(proxyTargetClass = true)
@Component("jobService")
public class JobService extends BasicService implements InitializingBean {

    @Override
    public void afterPropertiesSet() throws Exception {

        scheduler.init(new JobEngineConfig(kylinConfig), new ZookeeperJobLock());

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    scheduler.shutdown();
                } catch (SchedulerException e) {
                    logger.error("error occurred to shutdown scheduler", e);
                }
            }
        }));

3. 补丁分析

CVE-2022-24697

https://github.com/apache/kylin/pull/1811/files

upload successful

新增了ParameterFilter

CVE-2022-43396

https://github.com/apache/kylin/pull/2011/files

禁用了kylin.engine.spark-cmd 属性

4. CodeQL 复现

4.1 CodeQL 复现尝试

  1. 从api 到KylinConfig.override
    没问题
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
 * @kind path-problem
 */
import java
import semmle.code.java.dataflow.DataFlow
import semmle.code.java.dataflow.FlowSources
import semmle.code.java.dataflow.ExternalFlow
import InvokeSourceFlow::PathGraph


module InvokeSourceConfig implements DataFlow::ConfigSig {
  int fieldFlowBranchLimit() { result = 500 }

  predicate isSource(DataFlow::Node source) {
    source instanceof RemoteFlowSource
  }

  predicate isSink(DataFlow::Node sink) {
    // sinkNode(sink, "command-injection")
    exists( RefType rf|  rf.getASourceSupertype*().hasQualifiedName("org.apache.kylin.common", "KylinConfig") | 
      sink.asExpr().getFile() = rf.getFile()
    )
  }
}

module InvokeSourceFlow = TaintTracking::Global<InvokeSourceConfig>;

from InvokeSourceFlow::PathNode source, InvokeSourceFlow::PathNode sink
where InvokeSourceFlow::flowPath(source, sink)
select sink, source, sink, "$@->$@", source, source.toString(), sink, sink.toString()

upload successful

upload successful

  1. 从KylinConfig 到Sink
1

5. 思考

这个漏洞流程很复杂，涉及到3个线程，2个http 请求，想靠CodeQL 直接实现难度很大。提两个脑洞思路

  1. 寻找全局静态类实例，比如这个case 中的CubeManager
  2. 从http source 能够传播到CubeManager 中哪些field
  3. 从这些field 再作为source 点，能否传播至sink

上main中的全局类实例，也可以是

  • db
  • cache
  • 文件
  • 全局变量

实现起来应该是很难的了。。。

6. 参考

  1. https://xz.aliyun.com/news/11631
  2. https://lists.apache.org/thread/07mnn9c7o314wrhrwjr10w9j5s82voj4

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true