Skip to content

Commit 8a3b4f1

Browse files
loicgreffierloicgreffier
authored
Fix SSRF on schema registry client (#468)
* Attempt to fixing ssrf * Attempt fixing SSRF on schema registry client
1 parent a6e6c18 commit 8a3b4f1

File tree

2 files changed

+90
-43
lines changed

2 files changed

+90
-43
lines changed

‎src/main/java/com/michelin/ns4kafka/service/SchemaService.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -216,8 +216,10 @@ private Mono<List<String>> validateReferences(Namespace ns, Schema schema) {
216216
*/
217217
public Mono<Integer> register(Namespace namespace, Schema schema) {
218218
return schemaRegistryClient
219-
.register(namespace.getMetadata().getCluster(),
220-
schema.getMetadata().getName(), SchemaRequest.builder()
219+
.register(
220+
namespace.getMetadata().getCluster(),
221+
schema.getMetadata().getName(),
222+
SchemaRequest.builder()
221223
.schemaType(String.valueOf(schema.getSpec().getSchemaType()))
222224
.schema(schema.getSpec().getSchema())
223225
.references(schema.getSpec().getReferences())

‎src/main/java/com/michelin/ns4kafka/service/client/schema/SchemaRegistryClient.java

Lines changed: 86 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import jakarta.inject.Inject;
2525
import jakarta.inject.Singleton;
2626
import java.net.URI;
27+
import java.net.URLEncoder;
28+
import java.nio.charset.StandardCharsets;
2729
import java.util.Arrays;
2830
import java.util.List;
2931
import java.util.Optional;
@@ -71,9 +73,14 @@ public Flux<String> getSubjects(String kafkaCluster) {
7173
*/
7274
public Mono<SchemaResponse> getSubject(String kafkaCluster, String subject, String version) {
7375
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
76+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
77+
String encodedVersion = URLEncoder.encode(version, StandardCharsets.UTF_8);
78+
7479
HttpRequest<?> request = HttpRequest.GET(
75-
URI.create(StringUtils.prependUri(config.getUrl(), SUBJECTS + subject + VERSIONS + version)))
80+
URI.create(StringUtils.prependUri(config.getUrl(),
81+
SUBJECTS + encodedSubject + VERSIONS + encodedVersion)))
7682
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
83+
7784
return Mono.from(httpClient.retrieve(request, SchemaResponse.class))
7885
.onErrorResume(HttpClientResponseException.class,
7986
ex -> ex.getStatus().equals(HttpStatus.NOT_FOUND) ? Mono.empty() : Mono.error(ex));
@@ -88,15 +95,18 @@ public Mono<SchemaResponse> getSubject(String kafkaCluster, String subject, Stri
8895
*/
8996
public Flux<SchemaResponse> getAllSubjectVersions(String kafkaCluster, String subject) {
9097
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
98+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
99+
91100
HttpRequest<?> request = HttpRequest.GET(
92-
URI.create(StringUtils.prependUri(config.getUrl(), SUBJECTS + subject + "/versions")))
101+
URI.create(StringUtils.prependUri(config.getUrl(),
102+
SUBJECTS + encodedSubject + "/versions")))
93103
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
94104

95105
return Flux.from(httpClient.retrieve(request, Integer[].class))
96106
.flatMap(ids -> Flux.fromIterable(Arrays.asList(ids))
97107
.flatMap(id -> {
98-
HttpRequest<?> requestVersion = HttpRequest.GET(
99-
URI.create(StringUtils.prependUri(config.getUrl(), SUBJECTS + subject + VERSIONS + id)))
108+
HttpRequest<?> requestVersion = HttpRequest.GET(URI.create(StringUtils.prependUri(config.getUrl(),
109+
SUBJECTS + encodedSubject + VERSIONS + id)))
100110
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
101111

102112
return httpClient.retrieve(requestVersion, SchemaResponse.class);
@@ -115,10 +125,13 @@ public Flux<SchemaResponse> getAllSubjectVersions(String kafkaCluster, String su
115125
*/
116126
public Mono<SchemaResponse> register(String kafkaCluster, String subject, SchemaRequest body) {
117127
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
118-
HttpRequest<?> request =
119-
HttpRequest.POST(URI.create(StringUtils.prependUri(config.getUrl(), SUBJECTS + subject + "/versions")),
120-
body)
121-
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
128+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
129+
130+
HttpRequest<?> request = HttpRequest.POST(
131+
URI.create(StringUtils.prependUri(config.getUrl(),
132+
SUBJECTS + encodedSubject + "/versions")), body)
133+
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
134+
122135
return Mono.from(httpClient.retrieve(request, SchemaResponse.class));
123136
}
124137

@@ -132,9 +145,13 @@ public Mono<SchemaResponse> register(String kafkaCluster, String subject, Schema
132145
*/
133146
public Mono<Integer[]> deleteSubject(String kafkaCluster, String subject, boolean hardDelete) {
134147
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
148+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
149+
135150
MutableHttpRequest<?> request = HttpRequest.DELETE(
136-
URI.create(StringUtils.prependUri(config.getUrl(), SUBJECTS + subject + "?permanent=" + hardDelete)))
151+
URI.create(StringUtils.prependUri(config.getUrl(),
152+
SUBJECTS + encodedSubject + "?permanent=" + hardDelete)))
137153
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
154+
138155
return Mono.from(httpClient.retrieve(request, Integer[].class));
139156
}
140157

@@ -147,13 +164,16 @@ public Mono<Integer[]> deleteSubject(String kafkaCluster, String subject, boolea
147164
* @param hardDelete Should the subject be hard deleted or not
148165
* @return The version of the deleted subject
149166
*/
150-
public Mono<Integer> deleteSubjectVersion(String kafkaCluster, String subject, String version,
151-
boolean hardDelete) {
167+
public Mono<Integer> deleteSubjectVersion(String kafkaCluster, String subject, String version, boolean hardDelete) {
152168
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
169+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
170+
String encodedVersion = URLEncoder.encode(version, StandardCharsets.UTF_8);
171+
153172
MutableHttpRequest<?> request = HttpRequest.DELETE(
154-
URI.create(StringUtils.prependUri(config.getUrl(), SUBJECTS + subject + VERSIONS + version
155-
+ "?permanent=" + hardDelete)))
173+
URI.create(StringUtils.prependUri(config.getUrl(),
174+
SUBJECTS + encodedSubject + VERSIONS + encodedVersion + "?permanent=" + hardDelete)))
156175
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
176+
157177
return Mono.from(httpClient.retrieve(request, Integer.class));
158178
}
159179

@@ -165,14 +185,17 @@ public Mono<Integer> deleteSubjectVersion(String kafkaCluster, String subject, S
165185
* @param body The request
166186
* @return The schema compatibility validation
167187
*/
168-
public Mono<SchemaCompatibilityCheckResponse> validateSchemaCompatibility(String kafkaCluster, String subject,
188+
public Mono<SchemaCompatibilityCheckResponse> validateSchemaCompatibility(String kafkaCluster,
189+
String subject,
169190
SchemaRequest body) {
170191
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
171-
HttpRequest<?> request = HttpRequest.POST(URI.create(
172-
StringUtils.prependUri(config.getUrl(), "/compatibility/subjects/" + subject
173-
+ "/versions?verbose=true")),
174-
body)
192+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
193+
194+
HttpRequest<?> request = HttpRequest.POST(
195+
URI.create(StringUtils.prependUri(config.getUrl(),
196+
"/compatibility/subjects/" + encodedSubject + "/versions?verbose=true")), body)
175197
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
198+
176199
return Mono.from(httpClient.retrieve(request, SchemaCompatibilityCheckResponse.class))
177200
.onErrorResume(HttpClientResponseException.class,
178201
ex -> ex.getStatus().equals(HttpStatus.NOT_FOUND) ? Mono.empty() : Mono.error(ex));
@@ -186,12 +209,17 @@ public Mono<SchemaCompatibilityCheckResponse> validateSchemaCompatibility(String
186209
* @param body The schema compatibility request
187210
* @return The schema compatibility update
188211
*/
189-
public Mono<SchemaCompatibilityResponse> updateSubjectCompatibility(String kafkaCluster, String subject,
212+
public Mono<SchemaCompatibilityResponse> updateSubjectCompatibility(String kafkaCluster,
213+
String subject,
190214
SchemaCompatibilityRequest body) {
191215
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
192-
HttpRequest<?> request =
193-
HttpRequest.PUT(URI.create(StringUtils.prependUri(config.getUrl(), CONFIG + subject)), body)
194-
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
216+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
217+
218+
HttpRequest<?> request = HttpRequest.PUT(
219+
URI.create(StringUtils.prependUri(config.getUrl(),
220+
CONFIG + encodedSubject)), body)
221+
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
222+
195223
return Mono.from(httpClient.retrieve(request, SchemaCompatibilityResponse.class));
196224
}
197225

@@ -204,8 +232,13 @@ public Mono<SchemaCompatibilityResponse> updateSubjectCompatibility(String kafka
204232
*/
205233
public Mono<SchemaCompatibilityResponse> getCurrentCompatibilityBySubject(String kafkaCluster, String subject) {
206234
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
207-
HttpRequest<?> request = HttpRequest.GET(URI.create(StringUtils.prependUri(config.getUrl(), CONFIG + subject)))
235+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
236+
237+
HttpRequest<?> request = HttpRequest.GET(
238+
URI.create(StringUtils.prependUri(config.getUrl(),
239+
CONFIG + encodedSubject)))
208240
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
241+
209242
return Mono.from(httpClient.retrieve(request, SchemaCompatibilityResponse.class))
210243
.onErrorResume(HttpClientResponseException.class,
211244
ex -> ex.getStatus().equals(HttpStatus.NOT_FOUND) ? Mono.empty() : Mono.error(ex));
@@ -220,9 +253,13 @@ public Mono<SchemaCompatibilityResponse> getCurrentCompatibilityBySubject(String
220253
*/
221254
public Mono<SchemaCompatibilityResponse> deleteCurrentCompatibilityBySubject(String kafkaCluster, String subject) {
222255
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
223-
MutableHttpRequest<?> request =
224-
HttpRequest.DELETE(URI.create(StringUtils.prependUri(config.getUrl(), CONFIG + subject)))
225-
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
256+
String encodedSubject = URLEncoder.encode(subject, StandardCharsets.UTF_8);
257+
258+
MutableHttpRequest<?> request = HttpRequest.DELETE(
259+
URI.create(StringUtils.prependUri(config.getUrl(),
260+
CONFIG + encodedSubject)))
261+
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
262+
226263
return Mono.from(httpClient.retrieve(request, SchemaCompatibilityResponse.class));
227264
}
228265

@@ -235,11 +272,12 @@ public Mono<SchemaCompatibilityResponse> deleteCurrentCompatibilityBySubject(Str
235272
*/
236273
public Mono<List<TagTopicInfo>> associateTags(String kafkaCluster, List<TagTopicInfo> tagSpecs) {
237274
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
238-
HttpRequest<?> request = HttpRequest
239-
.POST(URI.create(StringUtils.prependUri(
240-
config.getUrl(),
275+
276+
HttpRequest<?> request = HttpRequest.POST(
277+
URI.create(StringUtils.prependUri(config.getUrl(),
241278
"/catalog/v1/entity/tags")), tagSpecs)
242279
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
280+
243281
return Mono.from(httpClient.retrieve(request, Argument.listOf(TagTopicInfo.class)));
244282
}
245283

@@ -252,9 +290,12 @@ public Mono<List<TagTopicInfo>> associateTags(String kafkaCluster, List<TagTopic
252290
*/
253291
public Mono<List<TagInfo>> createTags(String kafkaCluster, List<TagInfo> tags) {
254292
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
255-
HttpRequest<?> request = HttpRequest.POST(URI.create(StringUtils.prependUri(
256-
config.getUrl(), "/catalog/v1/types/tagdefs")), tags)
293+
294+
HttpRequest<?> request = HttpRequest.POST(
295+
URI.create(StringUtils.prependUri(config.getUrl(),
296+
"/catalog/v1/types/tagdefs")), tags)
257297
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
298+
258299
return Mono.from(httpClient.retrieve(request, Argument.listOf(TagInfo.class)));
259300
}
260301

@@ -268,11 +309,12 @@ public Mono<List<TagInfo>> createTags(String kafkaCluster, List<TagInfo> tags) {
268309
*/
269310
public Mono<HttpResponse<Void>> dissociateTag(String kafkaCluster, String entityName, String tagName) {
270311
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
271-
HttpRequest<?> request = HttpRequest
272-
.DELETE(URI.create(StringUtils.prependUri(
273-
config.getUrl(),
312+
313+
HttpRequest<?> request = HttpRequest.DELETE(
314+
URI.create(StringUtils.prependUri(config.getUrl(),
274315
"/catalog/v1/entity/type/kafka_topic/name/" + entityName + "/tags/" + tagName)))
275316
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
317+
276318
return Mono.from(httpClient.exchange(request, Void.class));
277319
}
278320

@@ -284,11 +326,12 @@ public Mono<HttpResponse<Void>> dissociateTag(String kafkaCluster, String entity
284326
*/
285327
public Mono<TopicListResponse> getTopicWithCatalogInfo(String kafkaCluster, int limit, int offset) {
286328
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
287-
HttpRequest<?> request = HttpRequest
288-
.GET(URI.create(StringUtils.prependUri(
289-
config.getUrl(), "/catalog/v1/search/basic?type=kafka_topic&limit="
290-
+ limit + "&offset=" + offset)))
329+
330+
HttpRequest<?> request = HttpRequest.GET(
331+
URI.create(StringUtils.prependUri(config.getUrl(),
332+
"/catalog/v1/search/basic?type=kafka_topic&limit=" + limit + "&offset=" + offset)))
291333
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
334+
292335
return Mono.from(httpClient.retrieve(request, TopicListResponse.class));
293336
}
294337

@@ -302,10 +345,12 @@ public Mono<TopicListResponse> getTopicWithCatalogInfo(String kafkaCluster, int
302345
public Mono<HttpResponse<TopicDescriptionUpdateResponse>> updateDescription(String kafkaCluster,
303346
TopicDescriptionUpdateBody body) {
304347
ManagedClusterProperties.SchemaRegistryProperties config = getSchemaRegistry(kafkaCluster);
305-
HttpRequest<?> request = HttpRequest
306-
.PUT(URI.create(StringUtils.prependUri(
307-
config.getUrl(), "/catalog/v1/entity")), body)
348+
349+
HttpRequest<?> request = HttpRequest.PUT(
350+
URI.create(StringUtils.prependUri(config.getUrl(),
351+
"/catalog/v1/entity")), body)
308352
.basicAuth(config.getBasicAuthUsername(), config.getBasicAuthPassword());
353+
309354
return Mono.from(httpClient.exchange(request, TopicDescriptionUpdateResponse.class));
310355
}
311356

0 commit comments

Comments
 (0)