Skip to content

Commit 18f5c36

Browse files
azhurazhur
authored
KAFKA-4090: Validate SSL connection in client for kafka 3.1.x
1 parent caac62e commit 18f5c36

File tree

12 files changed

+553
-111
lines changed

12 files changed

+553
-111
lines changed

‎.github/workflows/ci_pr.yml

+17
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
name: CI Pull request
2+
3+
on:
4+
pull_request:
5+
types: [ opened, reopened, synchronize ]
6+
7+
jobs:
8+
build:
9+
runs-on: ubuntu-latest
10+
steps:
11+
- uses: actions/checkout@v2
12+
- name: Set up JDK 11
13+
uses: actions/setup-java@v2
14+
with:
15+
java-version: '11'
16+
distribution: 'temurin'
17+
- run: ./gradlew clients:build

‎.github/workflows/release.yml

+32
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
name: Release
2+
3+
# Triggered when a draft released is "published" (not a draft anymore)
4+
on:
5+
release:
6+
types: [ published ]
7+
8+
jobs:
9+
release:
10+
runs-on: ubuntu-latest
11+
steps:
12+
- uses: actions/checkout@v2
13+
- name: Set up JDK 11
14+
uses: actions/setup-java@v2
15+
with:
16+
java-version: '11'
17+
distribution: 'temurin'
18+
- name: Set env
19+
run: echo "RELEASE_VERSION=${GITHUB_REF#refs/*/}" >> $GITHUB_ENV
20+
- name: Setup gradle.properties
21+
shell: bash
22+
env:
23+
USERNAME: ${{ github.actor }}
24+
PASSWORD: ${{ secrets.CONDUKTOR_BOT_TOKEN }}
25+
run: |
26+
echo "mavenUrl=https://maven.pkg.github.com/conduktor/kafka" >> ./gradle.properties
27+
echo "group=io.conduktor.kafka" >> ./gradle.properties
28+
echo "mavenUsername=$USERNAME" >> ./gradle.properties
29+
echo "mavenPassword=$PASSWORD" >> ./gradle.properties
30+
echo "version=$RELEASE_VERSION" >> ./gradle.properties
31+
echo "skipSigning=true" >> ./gradle.properties
32+
- run: ./gradlew clients:publish

‎checkstyle/checkstyle.xml

+2-1
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,10 @@
2727
<module name="Header">
2828
<property name="headerFile" value="${headerFile}" />
2929
</module>
30+
<module name="SuppressWarningsFilter" />
3031

3132
<module name="TreeWalker">
32-
33+
<module name="SuppressWarningsHolder" />
3334
<!-- code cleanup -->
3435
<module name="UnusedImports">
3536
<property name="processJavadoc" value="true" />

‎clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java

+133-70
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,13 @@
2323
import java.io.EOFException;
2424
import java.io.IOException;
2525
import java.nio.ByteBuffer;
26+
import java.nio.channels.ReadableByteChannel;
2627
import java.nio.channels.ScatteringByteChannel;
28+
import java.util.concurrent.atomic.AtomicInteger;
2729

2830
/**
29-
* A size delimited Receive that consists of a 4 byte network-ordered size N followed by N bytes of content
31+
* A size delimited Receive that consists of a 4 byte network-ordered size N
32+
* followed by N bytes of content.
3033
*/
3134
public class NetworkReceive implements Receive {
3235

@@ -36,129 +39,189 @@ public class NetworkReceive implements Receive {
3639
private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
3740

3841
private final String source;
39-
private final ByteBuffer size;
42+
private final ByteBuffer sizeBuf;
43+
private final ByteBuffer minBuf;
4044
private final int maxSize;
4145
private final MemoryPool memoryPool;
46+
private final AtomicInteger byteCount;
4247
private int requestedBufferSize = -1;
43-
private ByteBuffer buffer;
48+
private ByteBuffer payloadBuffer = null;
49+
private volatile ReadState readState = ReadState.READ_SIZE;
4450

51+
enum ReadState {
52+
READ_SIZE, VALIDATE_SIZE, ALLOCATE_BUFFER, READ_PAYLOAD, COMPLETE
53+
}
4554

46-
public NetworkReceive(String source, ByteBuffer buffer) {
47-
this.source = source;
48-
this.buffer = buffer;
49-
this.size = null;
50-
this.maxSize = UNLIMITED;
51-
this.memoryPool = MemoryPool.NONE;
55+
public NetworkReceive() {
56+
this(UNKNOWN_SOURCE);
5257
}
5358

5459
public NetworkReceive(String source) {
55-
this.source = source;
56-
this.size = ByteBuffer.allocate(4);
57-
this.buffer = null;
58-
this.maxSize = UNLIMITED;
59-
this.memoryPool = MemoryPool.NONE;
60+
this(UNLIMITED, source);
61+
}
62+
63+
public NetworkReceive(String source, ByteBuffer buffer) {
64+
this(source);
65+
this.payloadBuffer = buffer;
6066
}
6167

6268
public NetworkReceive(int maxSize, String source) {
63-
this.source = source;
64-
this.size = ByteBuffer.allocate(4);
65-
this.buffer = null;
66-
this.maxSize = maxSize;
67-
this.memoryPool = MemoryPool.NONE;
69+
this(maxSize, source, MemoryPool.NONE);
6870
}
6971

7072
public NetworkReceive(int maxSize, String source, MemoryPool memoryPool) {
7173
this.source = source;
72-
this.size = ByteBuffer.allocate(4);
73-
this.buffer = null;
7474
this.maxSize = maxSize;
7575
this.memoryPool = memoryPool;
76-
}
7776

78-
public NetworkReceive() {
79-
this(UNKNOWN_SOURCE);
80-
}
81-
82-
@Override
83-
public String source() {
84-
return source;
85-
}
86-
87-
@Override
88-
public boolean complete() {
89-
return !size.hasRemaining() && buffer != null && !buffer.hasRemaining();
77+
this.minBuf = (ByteBuffer) ByteBuffer.allocate(SslUtils.SSL_RECORD_HEADER_LENGTH).position(4);
78+
this.sizeBuf = (ByteBuffer) this.minBuf.duplicate().position(0).limit(4);
79+
this.byteCount = new AtomicInteger(0);
9080
}
9181

82+
@SuppressWarnings("fallthrough")
9283
public long readFrom(ScatteringByteChannel channel) throws IOException {
9384
int read = 0;
94-
if (size.hasRemaining()) {
95-
int bytesRead = channel.read(size);
96-
if (bytesRead < 0)
97-
throw new EOFException();
98-
read += bytesRead;
99-
if (!size.hasRemaining()) {
100-
size.rewind();
101-
int receiveSize = size.getInt();
102-
if (receiveSize < 0)
103-
throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
104-
if (maxSize != UNLIMITED && receiveSize > maxSize)
105-
throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
106-
requestedBufferSize = receiveSize; //may be 0 for some payloads (SASL)
107-
if (receiveSize == 0) {
108-
buffer = EMPTY_BUFFER;
85+
86+
switch (readState) {
87+
case READ_SIZE:
88+
read += readRequestedBufferSize(channel);
89+
if (this.sizeBuf.hasRemaining()) {
90+
break;
10991
}
110-
}
92+
this.readState = ReadState.VALIDATE_SIZE;
93+
/** FALLTHROUGH TO NEXT STATE */
94+
case VALIDATE_SIZE:
95+
if (this.requestedBufferSize != 0) {
96+
read += validateRequestedBufferSize(channel);
97+
if (this.minBuf.hasRemaining()) {
98+
break;
99+
}
100+
}
101+
this.readState = ReadState.ALLOCATE_BUFFER;
102+
/** FALLTHROUGH */
103+
case ALLOCATE_BUFFER:
104+
if (this.requestedBufferSize == 0) {
105+
this.payloadBuffer = EMPTY_BUFFER;
106+
} else {
107+
this.payloadBuffer = tryAllocateBuffer(this.requestedBufferSize);
108+
if (this.payloadBuffer == null) {
109+
break;
110+
} else {
111+
// Copy any bytes that were already consumed
112+
this.minBuf.position(this.sizeBuf.limit());
113+
this.payloadBuffer.put(this.minBuf);
114+
}
115+
}
116+
this.readState = ReadState.READ_PAYLOAD;
117+
/** FALLTHROUGH TO NEXT STATE */
118+
case READ_PAYLOAD:
119+
final int payloadRead = channel.read(payloadBuffer);
120+
if (payloadRead < 0)
121+
throw new EOFException();
122+
read += payloadRead;
123+
if (!this.payloadBuffer.hasRemaining()) {
124+
this.readState = ReadState.COMPLETE;
125+
}
126+
break;
127+
case COMPLETE:
128+
break;
111129
}
112-
if (buffer == null && requestedBufferSize != -1) { //we know the size we want but havent been able to allocate it yet
113-
buffer = memoryPool.tryAllocate(requestedBufferSize);
114-
if (buffer == null)
115-
log.trace("Broker low on memory - could not allocate buffer of size {} for source {}", requestedBufferSize, source);
130+
131+
this.byteCount.addAndGet(read);
132+
133+
return read;
134+
}
135+
136+
private int validateRequestedBufferSize(final ScatteringByteChannel channel)
137+
throws IOException {
138+
int minRead = channel.read(this.minBuf);
139+
if (minRead < 0) {
140+
throw new EOFException();
116141
}
117-
if (buffer != null) {
118-
int bytesRead = channel.read(buffer);
119-
if (bytesRead < 0)
120-
throw new EOFException();
121-
read += bytesRead;
142+
if (!this.minBuf.hasRemaining()) {
143+
final boolean isEncrypted =
144+
SslUtils.isEncrypted((ByteBuffer) this.minBuf.duplicate().rewind());
145+
if (isEncrypted) {
146+
throw new InvalidReceiveException(
147+
"Recieved an unexpected SSL packet from the server. "
148+
+ "Please ensure the client is properly configured with SSL enabled.");
149+
}
150+
if (this.requestedBufferSize < 0)
151+
throw new InvalidReceiveException(
152+
"Invalid receive (size = " + this.requestedBufferSize + ")");
153+
if (maxSize != UNLIMITED && this.requestedBufferSize > maxSize)
154+
throw new InvalidReceiveException("Invalid receive (size = "
155+
+ this.requestedBufferSize + " larger than " + maxSize + ")");
122156
}
123157

124-
return read;
158+
return minRead;
159+
}
160+
161+
private ByteBuffer tryAllocateBuffer(final int bufSize) {
162+
final ByteBuffer bb = memoryPool.tryAllocate(bufSize);
163+
if (bb == null) {
164+
log.trace("Broker low on memory - could not allocate buffer of size {} for source {}",
165+
requestedBufferSize, source);
166+
}
167+
return bb;
168+
}
169+
170+
private int readRequestedBufferSize(final ReadableByteChannel channel) throws IOException {
171+
final int sizeRead = channel.read(sizeBuf);
172+
if (sizeRead < 0) {
173+
throw new EOFException();
174+
}
175+
if (sizeBuf.hasRemaining()) {
176+
return sizeRead;
177+
}
178+
sizeBuf.rewind();
179+
this.requestedBufferSize = sizeBuf.getInt();
180+
return sizeRead;
125181
}
126182

127183
@Override
128184
public boolean requiredMemoryAmountKnown() {
129-
return requestedBufferSize != -1;
185+
return this.readState.ordinal() > ReadState.VALIDATE_SIZE.ordinal();
130186
}
131187

132188
@Override
133189
public boolean memoryAllocated() {
134-
return buffer != null;
190+
return this.readState.ordinal() >= ReadState.READ_PAYLOAD.ordinal();
135191
}
136192

193+
@Override
194+
public boolean complete() {
195+
return this.readState == ReadState.COMPLETE;
196+
}
137197

138198
@Override
139199
public void close() throws IOException {
140-
if (buffer != null && buffer != EMPTY_BUFFER) {
141-
memoryPool.release(buffer);
142-
buffer = null;
200+
if (payloadBuffer != null && payloadBuffer != EMPTY_BUFFER) {
201+
memoryPool.release(payloadBuffer);
202+
payloadBuffer = null;
143203
}
144204
}
145205

206+
@Override
207+
public String source() {
208+
return source;
209+
}
210+
146211
public ByteBuffer payload() {
147-
return this.buffer;
212+
return this.payloadBuffer;
148213
}
149214

150215
public int bytesRead() {
151-
if (buffer == null)
152-
return size.position();
153-
return buffer.position() + size.position();
216+
return this.byteCount.get();
154217
}
155218

156219
/**
157220
* Returns the total size of the receive including payload and size buffer
158221
* for use in metrics. This is consistent with {@link NetworkSend#size()}
159222
*/
160223
public int size() {
161-
return payload().limit() + size.limit();
224+
return payload().limit() + sizeBuf.limit();
162225
}
163226

164227
}

0 commit comments

Comments
 (0)