[原创]Java代码审计的手法和高级审计工具链-WEB安全-看雪-安全社区|安全招聘|kanxue.com

[原创]Java代码审计的手法和高级审计工具链

发表于: 2025-4-3 10:33 690

[原创]Java代码审计的手法和高级审计工具链

Hrlies

2025-4-3 10:33

690

建议采用分层审计流程：

典型工具链组合：

通过多维度的交叉验证，可有效发现传统单一方法难以检测的深层逻辑漏洞（如条件竞争、业务链漏洞等）。实际审计中需要结合具体技术栈特点，例如对区块链智能合约需重点检查重入漏洞，云原生应用需审计IAM权限配置。

CodeQL高级配置

Semgrep定制规则

Burp Suite流量重放

IAST探针部署

# 匹配PHP中危险函数调用

pattern = r'(system|exec|shell_exec|passthru)\s*\('
# 匹配Java反序列化点

r'\.readObject\s*\('

# 匹配PHP中危险函数调用

pattern = r'(system|exec|shell_exec|passthru)\s*\('

# 匹配Java反序列化点

r'\.readObject\s*\('

String userInput = request.getParameter("data");

// 未过滤直接拼接

String query = "SELECT * FROM users WHERE id=" + userInput; 

// SQL注入漏洞点
stmt.execute(query);

String userInput = request.getParameter("data");

// 未过滤直接拼接

String query = "SELECT * FROM users WHERE id=" + userInput;

// SQL注入漏洞点

stmt.execute(query);

def delete_file(request):

    if request.user.is_admin:  # 权限校验

        pass

    # 缺少else分支，未授权可继续执行

    os.remove(request.file)  # 未鉴权执行危险操作

def delete_file(request):

if request.user.is_admin: # 权限校验

pass

# 缺少else分支，未授权可继续执行

os.remove(request.file) # 未鉴权执行危险操作

// Hook浏览器localStorage

let originalSetItem = Storage.prototype.setItem;

Storage.prototype.setItem = function(key, value) {

    console.log(`存储敏感数据: ${key}=${value}`);

    originalSetItem.apply(this, [key, value]);
};

// Hook浏览器localStorage

let originalSetItem = Storage.prototype.setItem;

Storage.prototype.setItem = function(key, value) {

console.log(`存储敏感数据: ${key}=${value}`);

originalSetItem.apply(this, [key, value]);

};

// 函数断点观察参数传递

(gdb) break validateUser

(gdb) watch *(int*)0x7fffffffda9c  // 监控权限标志位

// 函数断点观察参数传递

(gdb) break validateUser

(gdb) watch *(int*)0x7fffffffda9c // 监控权限标志位

void processPayment(Order order) {

    if (order.amount > 0) {

        deductBalance(order.user, order.amount); 

        // 未校验重复提交，可导致重复扣款

    }
}

void processPayment(Order order) {

if (order.amount > 0) {

deductBalance(order.user, order.amount);

// 未校验重复提交，可导致重复扣款

}

# 检测使用不安全的哈希算法

if "md5(" in line or "sha1(" in line:

    report_vulnerability(line_num, "Weak hash algorithm")

# 检测使用不安全的哈希算法

if "md5(" in line or "sha1(" in line:

report_vulnerability(line_num, "Weak hash algorithm")

# 使用OWASP Dependency-Check扫描

dependency-check.sh --project MyApp --scan ./lib
# 输出包含CVE-2021-44228（Log4j）

# 使用OWASP Dependency-Check扫描

dependency-check.sh --project MyApp --scan ./lib

# 输出包含CVE-2021-44228（Log4j）

security:

  enabled: false  # 错误禁用安全模块

  cors:

    allowed-origins: "*"  # 不安全的CORS配置

security:

enabled: false # 错误禁用安全模块

cors:

allowed-origins: "*" # 不安全的CORS配置

from Call call, Expr arg

where call.getTarget().hasName("executeQuery")

  and arg = call.getArgument(0)

  and not exists(SanitizationCheck arg)

select call, "Potential SQL injection"

from Call call, Expr arg

where call.getTarget().hasName("executeQuery")

and arg = call.getArgument(0)

and not exists(SanitizationCheck arg)

select call, "Potential SQL injection"

dangerous_commands = ["Runtime.exec", "ProcessBuilder"]

sinks = ["system", "popen", "exec"]

dangerous_commands = ["Runtime.exec", "ProcessBuilder"]

sinks = ["system", "popen", "exec"]

// 检测不安全的字符串操作

strcpy(dest, src);  // 无长度检查

sprintf(buffer, "%s", input); // 未限制长度

// 检测不安全的字符串操作

strcpy(dest, src); // 无长度检查

sprintf(buffer, "%s", input); // 未限制长度

@PostMapping("/upload")

public String handleUpload(@RequestParam MultipartFile file) {

    // 文件上传处理逻辑
}

@PostMapping("/upload")

public String handleUpload(@RequestParam MultipartFile file) {

// 文件上传处理逻辑

}

攻击复杂度: Low
影响范围: High
利用难度: Medium

CVSS 3.1: 8.1 (High)

攻击复杂度: Low

影响范围: High

利用难度: Medium

CVSS 3.1: 8.1 (High)

graph TB

    A[数据采集层] --> B[静态分析层]

    A --> C[动态分析层]

    A --> D[交互式分析层]

    B --> E[规则引擎]

    C --> E

    D --> E

    E --> F[漏洞管理平台]

graph TB

A[数据采集层] --> B[静态分析层]

A --> C[动态分析层]

A --> D[交互式分析层]

B --> E[规则引擎]

C --> E

D --> E

E --> F[漏洞管理平台]

层级	工具/框架	主要能力	Java适用性
静态分析	CodeQL/Semgrep	语义级漏洞扫描	★★★★★
	FindSecBugs	基础漏洞模式检测	★★★★☆
动态分析	Burp Suite Pro	流量Hook/漏洞验证	★★★★☆
	OWASP ZAP API	自动化API测试	★★★☆☆
IAST	Contrast Security	运行时插桩检测	★★★★★
基准库	CWE/SANS Top25	漏洞模式库	★★★★☆
集成平台	SonarQube + Jenkins	流水线集成	★★★★☆

# 创建Java数据库

codeql database create java-db --language=java --command="mvn clean install"
 
# 自定义安全规则（示例：检测反序列化漏洞）

import java
 
from MethodAccess method

where method.getDeclaringType().hasQualifiedName("java.io", "ObjectInputStream") 

  and method.getName() = "readObject"

select method, "潜在反序列化点，需校验输入来源"

# 创建Java数据库

codeql database create java-db --language=java --command="mvn clean install"

# 自定义安全规则（示例：检测反序列化漏洞）

import java

from MethodAccess method

where method.getDeclaringType().hasQualifiedName("java.io", "ObjectInputStream")

and method.getName() = "readObject"

select method, "潜在反序列化点，需校验输入来源"

rules:

- id: unsafe-reflection

  pattern: |

    Class.forName($CLASS).getMethod($METHOD).invoke(...)

  message: 发现危险反射调用

  languages: [java]

  severity: WARNING

rules:

- id: unsafe-reflection

pattern: |

Class.forName($CLASS).getMethod($METHOD).invoke(...)

message: 发现危险反射调用

languages: [java]

severity: WARNING

# 使用Burp API自动化测试（Python示例）

from burp import IBurpExtender

from burp import IHttpListener
 
class BurpExtender(IBurpExtender, IHttpListener):

    def processHttpMessage(self, toolFlag, messageIsRequest, message):

        if not messageIsRequest:

            response = message.getResponse()

            if b"SQL syntax error" in response:

                self._callbacks.issueAlert("发现SQL注入漏洞")

# 使用Burp API自动化测试（Python示例）

from burp import IBurpExtender

from burp import IHttpListener

class BurpExtender(IBurpExtender, IHttpListener):

def processHttpMessage(self, toolFlag, messageIsRequest, message):

if not messageIsRequest:

response = message.getResponse()

if b"SQL syntax error" in response:

self._callbacks.issueAlert("发现SQL注入漏洞")

<!-- Contrast Security Maven配置 -->
<dependency>

    <groupId>com.contrastsecurity</groupId>

    <artifactId>contrast-agent</artifactId>

    <version>3.9.5</version>

</dependency>
 
# 启动参数

java -javaagent:contrast-agent.jar -Dcontrast.server=prod-1 -jar app.jar

<dependency>

<groupId>com.contrastsecurity</groupId>

<artifactId>contrast-agent</artifactId>

<version>3.9.5</version>

</dependency>

# 启动参数

java -javaagent:contrast-agent.jar -Dcontrast.server=prod-1 -jar app.jar

# 数据流追踪引擎架构

class TaintTracker:

    def __init__(self):

        self.sources = ["HttpServletRequest.getParameter", "JdbcTemplate.query"]  # 污染源

        self.sinks = ["Runtime.exec", "executeQuery"]  # 危险接收点

        self.propagation_rules = {}  # 传播规则
 
    def analyze_method(self, method):

        for inst in method.bytecode:

            if inst in self.sources:

                mark_as_tainted(inst.destination)

            elif inst in self.sinks and is_tainted(inst.source):

                report_vulnerability()

# 数据流追踪引擎架构

class TaintTracker:

def __init__(self):

self.sources = ["HttpServletRequest.getParameter", "JdbcTemplate.query"] # 污染源

self.sinks = ["Runtime.exec", "executeQuery"] # 危险接收点

self.propagation_rules = {} # 传播规则

def analyze_method(self, method):

for inst in method.bytecode:

if inst in self.sources:

mark_as_tainted(inst.destination)

elif inst in self.sinks and is_tainted(inst.source):

report_vulnerability()

// 基于AST的漏洞模式发现（使用Eclipse JDT）

CompilationUnit cu = parse(sourceCode);
cu.accept(new ASTVisitor() {

    public boolean visit(MethodInvocation node) {

        if (node.getName().toString().equals("executeQuery")) {

            Expression arg = node.arguments().get(0);

            if (arg instanceof InfixExpression) {

                report("SQL拼接漏洞", node.getStartPosition());

            }

        }

        return true;

    }
});

// 基于AST的漏洞模式发现（使用Eclipse JDT）

CompilationUnit cu = parse(sourceCode);

cu.accept(new ASTVisitor() {

public boolean visit(MethodInvocation node) {

if (node.getName().toString().equals("executeQuery")) {

Expression arg = node.arguments().get(0);

if (arg instanceof InfixExpression) {

report("SQL拼接漏洞", node.getStartPosition());

}

return true;

}

});

-- 漏洞数据库设计
CREATE TABLE vuln_reports (

    id INT PRIMARY KEY,

    tool VARCHAR(20),      -- 来源工具

    category VARCHAR(50),  -- 漏洞类型

    severity INT,          -- 危险等级

    path VARCHAR(200),     -- 代码路径

    hash CHAR(64)          -- 唯一标识
);
 
-- 去重算法伪代码

def deduplicate(reports):

    seen = set()

    for report in reports:

        key = hash(report['path'] + report['category'])

        if key not in seen:

            seen.add(key)

            yield report

-- 漏洞数据库设计

CREATE TABLE vuln_reports (

id INT PRIMARY KEY,

tool VARCHAR(20), -- 来源工具

category VARCHAR(50), -- 漏洞类型

severity INT, -- 危险等级

path VARCHAR(200), -- 代码路径

hash CHAR(64) -- 唯一标识

);

-- 去重算法伪代码

def deduplicate(reports):

seen = set()

for report in reports:

key = hash(report['path'] + report['category'])

if key not in seen:

seen.add(key)

yield report

pipeline {

    agent any

    stages {

        stage('Static Analysis') {

            steps {

                sh 'codeql analyze --format=sarif-latest --output=results.sarif'

                semgrep --config=p/security-audit --json > semgrep.json

            }

        }

        stage('Dynamic Test') {

            steps {

                zap-baseline.py -t http://localhost:8080 -r report.html

            }

        }

        stage('Report') {

            steps {

                vuln-aggregator --input=results.sarif,semgrep.json --output=merged.pdf

            }

        }

    }

    post {

        always {

            archiveArtifacts artifacts: '**/*.html,**/*.pdf'

        }

    }
}