CodeQL 提升篇 - 高级技巧与实践指南

一、基础功能增强

1. 编译闭源项目创建数据库

• 使用工具: codeql_compile
• 解决闭源项目无法直接创建CodeQL数据库的问题

2. 历史查询功能

• 在VSCode左侧的QUERY HISTORY中:
  • 点击切换历史查询内容
  • 右键可进行结果比对等功能

3. 查看AST

• 操作步骤:
  1. 在VSCode左侧选中Java文件
  2. 点击"View AST"查看抽象语法树
  3. 点击Java文件中的类/方法会自动定位到AST对应节点

4. 快速查询

• 在编写谓词上方有快速查询按钮
• 点击可立即查询当前谓词的结果

二、核心语法技巧

1. 获取具体QL类型

from Expr e, Callable c
where e.getEnclosingCallable() = c
select e, e.getAQlClass()

2. 范围缩小优化

优化前(性能差):

override predicate isSink(DataFlow::Node sink) {
  sink.asExpr().getParent() instanceof ReturnStmt
}

优化后(添加限定条件):

override predicate isSink(DataFlow::Node sink) {
  sink.asExpr().getParent() instanceof ReturnStmt and
  sink.asExpr().getEnclosingCallable().hasName("xxxxx")
}

3. 常用规则模板

方法参数作为source

override predicate isSource(DataFlow::Node source) {
  exists(Parameter p | 
    p.getCallable().hasName("readValue") and
    source.asParameter() = p and
    source.asParameter().getPosition() = 0 and
    p.getCallable().getDeclaringType().hasQualifiedName("com.service.impl", "xxxxx")
  )
}

实例参数作为source

override predicate isSource(DataFlow::Node source) {
  exists(ClassInstanceExpr ma |
    source.asExpr() = ma.getAnArgument() and
    ma.getTypeName().toString() = "X1" and
    ma.getCaller().hasName("Caller")
  )
}

4. 调用路径追踪

import java

class StartMethod extends Method {
  StartMethod() { getName() = "main" }
}

class TargetMethod extends Method {
  TargetMethod() { getName() = "vulMain" }
}

query predicate edges(Method a, Method b) { a.calls(b) }

from TargetMethod end, StartMethod entryPoint
where edges+(entryPoint, end)
select end, entryPoint, end, "Found a path from start to target."

5. 接口实现检测

class JsonInterface extends Interface {
  JsonInterface() {
    this.hasQualifiedName("com.alibaba.fastjson", "JSONStreamAware")
  }
  
  Method getJsonMethod() {
    result.getDeclaringType() = this
  }
}

class CMethod extends Method {
  CMethod() {
    this.overridesOrInstantiates*(any(JsonInterface i).getJsonMethod())
  }
}

from CMethod m select m, m.getDeclaringType()

三、查询结果处理

1. 查询类型与元数据

• 普通查询:

  • 元数据: @kind problem
  • 格式: select element, string
  • 禁止导入PathGraph相关

• 路径查询:

  • 元数据: @kind path-problem
  • 格式: select element, source, sink, string

2. 常见错误处理

错误示例:

Showing raw results instead of interpreted ones due to an error...
Expected result pattern(s) are not present for problem query...

解决方案:

• 确保查询类型与元数据匹配
• 检查select语句格式是否正确
• 避免在普通查询中使用PathGraph导入

四、数据流中断处理(AdditionalTaintStep)

1. Getter/Setter处理

class GetSetTaintStep extends TaintTracking::AdditionalTaintStep {
  override predicate step(DataFlow::Node src, DataFlow::Node sink) {
    exists(MethodAccess ma |
      (ma.getMethod() instanceof GetterMethod or
       ma.getMethod() instanceof SetterMethod or
       ma.getMethod().getName().matches("get%") or
       ma.getMethod().getName().matches("set%")) and
      src.asExpr() = ma.getQualifier() and
      sink.asExpr() = ma
    )
  }
}

2. MyBatis Mapper处理

class MapperTaintStep extends TaintTracking::AdditionalTaintStep {
  override predicate step(DataFlow::Node src, DataFlow::Node sink) {
    exists(MethodAccess ma |
      (ma.getQualifier().getType().getName().matches("%Dao") or
       ma.getQualifier().getType().getName().matches("%Mapper")) and
      src.asExpr() = ma.getAnArgument() and
      sink.asExpr() = ma
    )
  }
}

3. 方法参数污染传播

class SrcTaintStep extends TaintTracking::AdditionalTaintStep {
  override predicate step(DataFlow::Node src, DataFlow::Node sink) {
    exists(MethodAccess ma |
      (ma.getMethod() instanceof SetterMethod or
       ma.getMethod().getName().matches("set%")) and
      src.asExpr() = ma.getAnArgument() and
      sink.asExpr() = ma.getQualifier()
    )
  }
}

4. 实例化对象处理

class InstanceTaintStep extends TaintTracking::AdditionalTaintStep {
  override predicate step(DataFlow::Node src, DataFlow::Node sink) {
    exists(ClassInstanceExpr cie |
      // cie.getTypeName().toString() = "UploadFile" // 可添加特定类型过滤
      src.asExpr() = cie.getAnArgument() and
      sink.asExpr() = cie
    )
  }
}

五、部分流分析(Partial Flow)

1. 基本使用

import DataFlow::PartialPathGraph

class MyTaintTrackingConfiguration extends TaintTracking::Configuration {
  override int explorationLimit() { result = 5 }
}

from MyTaintTrackingConfiguration conf, 
     DataFlow::PartialPathNode source, 
     DataFlow::PartialPathNode sink
where conf.hasPartialFlow(source, sink, _)
select sink, source, sink, "Partial flow from unsanitized user data"

2. 调试技巧

1. 分段检查: 将长调用链分成小段检查
2. 精确source: 使用确定的单个source减少输出
3. 使用sanitizer: 清洗掉无关数据

六、路径注入检测实战

1. 官方规则分析

• Source: 使用RemoteFlowSource(常见用户输入源)
• Sink: 文件操作方法的文件名参数

  override predicate isSink(DataFlow::Node sink) {
    exists(Expr e | 
      e = sink.asExpr() | 
      e = any(PathCreation p).getAnInput() and 
      not guarded(e)
    )
  }
  

2. Guarded谓词分析

private predicate guarded(Expr e) {
  exists(ConditionBlock cb, Expr c |
    exists(PathCreation p | e = p.getAnInput()) and
    cb.getCondition().getAChildExpr*() = c and
    c = e.getVariable().getAnAccess() and
    cb.controls(e.getBasicBlock(), true) and
    not inWeakCheck(c)
  )
}

3. 清洗条件

override predicate isSanitizer(DataFlow::Node node) {
  exists(Type t | t = node.getType() | 
    t instanceof BoxedType or 
    t instanceof PrimitiveType
  )
}

override predicate isSanitizerGuard(DataFlow::BarrierGuard guard) {
  guard instanceof ContainsDotDotSanitizer
}

4. 真实场景应用

class TaintedPathConfig extends TaintTracking::Configuration {
  TaintedPathConfig() { this = "TaintedPathConfig" }
  
  override predicate isSource(DataFlow::Node source) {
    exists(Method m, Parameter p |
      m.getAnAnnotation().getType().hasQualifiedName(
        "org.springframework.web.bind.annotation", "RequestMapping") and
      m.hasAnnotation() and
      m.getAParameter() = p and
      source.asParameter() = p and
      p.getType().hasName("HttpServletRequest")
    )
  }
  
  override predicate isSink(DataFlow::Node sink) {
    exists(Method m, Parameter p |
      m.hasName("uploadFile") and
      m.getDeclaringType().hasQualifiedName(
        "org.xxxx.core.common.dao.impl", "xxxxx") and
      m.getAParameter() = p and
      sink.asParameter() = p and
      p.getType().hasName("UploadFile")
    )
  }
}

七、最佳实践总结

1. 性能优化:

   • 尽量缩小查询范围
   • 添加具体的限定条件

2. 数据流处理:

   • 识别常见中断场景
   • 合理使用AdditionalTaintStep
   • 掌握Partial Flow调试技巧

3. 规则编写:

   • 区分普通查询和路径查询
   • 正确处理元数据
   • 结合实际业务场景定制规则

4. 安全检测:

   • 理解官方规则设计思路
   • 根据实际项目调整source/sink定义
   • 合理设置sanitizer减少误报