Skip to content

Commit 55250e7

Browse files
seonoseono
committed
Merge remote-tracking branch 'real-origin/main' into issues/344
2 parents 8c4d0ff + 9ea1a4e commit 55250e7

File tree

20 files changed

+1304
-35
lines changed

20 files changed

+1304
-35
lines changed

‎api/build.gradle

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,15 @@ dependencies {
6262
implementation libs.netty.common
6363
implementation libs.netty.handler
6464

65+
66+
// Google Managed Service for Kafka IAM support
67+
implementation (libs.google.managed.kafka.login.handler) {
68+
exclude group: 'com.google.oauth-client', module: 'google-oauth-client'
69+
}
70+
implementation (libs.google.oauth.client) {
71+
because("CVE Fix: It is excluded above because of a vulnerability")
72+
}
73+
6574
// Annotation processors
6675
implementation libs.lombok
6776
implementation libs.mapstruct

‎api/src/main/java/io/kafbat/ui/emitter/MessageFilters.java

Lines changed: 30 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import com.fasterxml.jackson.databind.ObjectMapper;
88
import com.google.common.collect.ImmutableCollection;
99
import com.google.common.collect.ImmutableSet;
10+
import com.google.protobuf.NullValue;
1011
import dev.cel.common.CelAbstractSyntaxTree;
1112
import dev.cel.common.CelOptions;
1213
import dev.cel.common.CelValidationException;
@@ -26,6 +27,7 @@
2627
import io.kafbat.ui.exception.CelException;
2728
import io.kafbat.ui.model.TopicMessageDTO;
2829
import java.util.HashMap;
30+
import java.util.LinkedHashMap;
2931
import java.util.Map;
3032
import java.util.Objects;
3133
import java.util.Optional;
@@ -38,11 +40,13 @@
3840
@Slf4j
3941
@UtilityClass
4042
public class MessageFilters {
43+
4144
private static final String CEL_RECORD_VAR_NAME = "record";
4245
private static final String CEL_RECORD_TYPE_NAME = TopicMessageDTO.class.getSimpleName();
4346

4447
private static final CelCompiler CEL_COMPILER = createCompiler();
4548
private static final CelRuntime CEL_RUNTIME = createRuntime();
49+
private static final Object CELL_NULL_VALUE = NullValue.NULL_VALUE;
4650

4751
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
4852

@@ -188,10 +192,34 @@ private static Object parseToJsonOrReturnAsIs(@Nullable String str) {
188192
}
189193

190194
try {
191-
return OBJECT_MAPPER.readValue(str, new TypeReference<Map<String, Object>>() {
192-
});
195+
//@formatter:off
196+
var map = OBJECT_MAPPER.readValue(str, new TypeReference<Map<String, Object>>() {});
197+
//@formatter:on
198+
return replaceCelNulls(map);
193199
} catch (JsonProcessingException e) {
194200
return str;
195201
}
196202
}
203+
204+
@SuppressWarnings("unchecked")
205+
private static Map<String, Object> replaceCelNulls(Map<String, Object> map) {
206+
var result = new LinkedHashMap<String, Object>();
207+
208+
for (var entry : map.entrySet()) {
209+
String key = entry.getKey();
210+
Object value = entry.getValue();
211+
212+
if (value == null) {
213+
result.put(key, CELL_NULL_VALUE);
214+
} else if (value instanceof Map<?, ?>) {
215+
var inner = (Map<String, Object>) value;
216+
result.put(key, replaceCelNulls(inner));
217+
} else {
218+
result.put(key, value);
219+
}
220+
}
221+
222+
return result;
223+
}
224+
197225
}

‎api/src/test/java/io/kafbat/ui/emitter/MessageFiltersTest.java

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,6 +201,18 @@ void filterSpeedIsAtLeast5kPerSec() {
201201
assertThat(took).isLessThan(1000);
202202
assertThat(matched).isPositive();
203203
}
204+
205+
@Test
206+
void nullFiltering() {
207+
String msg = "{ \"field\": { \"inner\": null } }";
208+
209+
var f = celScriptFilter("record.value.field.inner == null");
210+
assertTrue(f.test(msg().content(msg)));
211+
212+
f = celScriptFilter("record.value.field.inner != null");
213+
assertFalse(f.test(msg().content(msg)));
214+
}
215+
204216
}
205217

206218
@Test

‎api/src/test/java/io/kafbat/ui/service/rbac/AccessControlServiceRbacDisabledTest.java

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
package io.kafbat.ui.service.rbac;
2+
3+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.CONNECT_NAME;
4+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.CONSUMER_GROUP_NAME;
5+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.DEV_ROLE;
6+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.PROD_CLUSTER;
7+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.SCHEMA_NAME;
8+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.TOPIC_NAME;
9+
import static io.kafbat.ui.service.rbac.MockedRbacUtils.getAccessContext;
10+
import static org.assertj.core.api.Assertions.assertThat;
11+
import static org.mockito.Mockito.when;
12+
13+
import io.kafbat.ui.AbstractIntegrationTest;
14+
import io.kafbat.ui.config.auth.RbacUser;
15+
import io.kafbat.ui.model.ClusterDTO;
16+
import io.kafbat.ui.model.ConnectDTO;
17+
import io.kafbat.ui.model.InternalTopic;
18+
import io.kafbat.ui.model.rbac.AccessContext;
19+
import io.kafbat.ui.model.rbac.Role;
20+
import java.util.List;
21+
import org.junit.jupiter.api.BeforeEach;
22+
import org.junit.jupiter.api.Test;
23+
import org.mockito.Mock;
24+
import org.mockito.MockedStatic;
25+
import org.mockito.Mockito;
26+
import org.springframework.beans.factory.annotation.Autowired;
27+
import org.springframework.security.core.Authentication;
28+
import org.springframework.security.core.context.ReactiveSecurityContextHolder;
29+
import org.springframework.security.core.context.SecurityContext;
30+
import org.springframework.test.annotation.DirtiesContext;
31+
import reactor.core.publisher.Mono;
32+
import reactor.test.StepVerifier;
33+
34+
/**
35+
* Test cases for AccessControlService when RBAC is disabled.
36+
* Using PROD cluster and user DEV role for all tests.
37+
*/
38+
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
39+
class AccessControlServiceRbacDisabledTest extends AbstractIntegrationTest {
40+
41+
@Autowired
42+
AccessControlService accessControlService;
43+
44+
@Mock
45+
SecurityContext securityContext;
46+
47+
@Mock
48+
Authentication authentication;
49+
50+
@Mock
51+
RbacUser user;
52+
53+
@BeforeEach
54+
void setUp() {
55+
// Mock security context
56+
when(securityContext.getAuthentication()).thenReturn(authentication);
57+
when(authentication.getPrincipal()).thenReturn(user);
58+
}
59+
60+
public void withSecurityContext(Runnable runnable) {
61+
try (MockedStatic<ReactiveSecurityContextHolder> ctxHolder = Mockito.mockStatic(
62+
ReactiveSecurityContextHolder.class)) {
63+
// Mock static method to get security context
64+
ctxHolder.when(ReactiveSecurityContextHolder::getContext).thenReturn(Mono.just(securityContext));
65+
runnable.run();
66+
}
67+
}
68+
69+
@Test
70+
void validateAccess() {
71+
withSecurityContext(() -> {
72+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
73+
AccessContext context = getAccessContext(PROD_CLUSTER, true);
74+
Mono<Void> validateAccessMono = accessControlService.validateAccess(context);
75+
StepVerifier.create(validateAccessMono)
76+
.expectComplete()
77+
.verify();
78+
});
79+
}
80+
81+
@Test
82+
void isClusterAccessible() {
83+
withSecurityContext(() -> {
84+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
85+
ClusterDTO clusterDto = new ClusterDTO();
86+
clusterDto.setName(PROD_CLUSTER);
87+
Mono<Boolean> clusterAccessibleMono = accessControlService.isClusterAccessible(clusterDto);
88+
StepVerifier.create(clusterAccessibleMono)
89+
.expectNext(true)
90+
.expectComplete()
91+
.verify();
92+
});
93+
}
94+
95+
@Test
96+
void filterViewableTopics() {
97+
withSecurityContext(() -> {
98+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
99+
List<InternalTopic> topics = List.of(
100+
InternalTopic.builder()
101+
.name(TOPIC_NAME)
102+
.build()
103+
);
104+
Mono<List<InternalTopic>> filterTopicsMono = accessControlService.filterViewableTopics(topics, PROD_CLUSTER);
105+
StepVerifier.create(filterTopicsMono)
106+
.expectNextMatches(responseTopics -> responseTopics.stream().anyMatch(t -> t.getName().equals(TOPIC_NAME)))
107+
.expectComplete()
108+
.verify();
109+
});
110+
}
111+
112+
@Test
113+
void isConsumerGroupAccessible() {
114+
withSecurityContext(() -> {
115+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
116+
Mono<Boolean> consumerGroupAccessibleMono =
117+
accessControlService.isConsumerGroupAccessible(CONSUMER_GROUP_NAME, PROD_CLUSTER);
118+
StepVerifier.create(consumerGroupAccessibleMono)
119+
.expectNext(true)
120+
.expectComplete()
121+
.verify();
122+
});
123+
}
124+
125+
@Test
126+
void isSchemaAccessible() {
127+
withSecurityContext(() -> {
128+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
129+
Mono<Boolean> consumerGroupAccessibleMono =
130+
accessControlService.isSchemaAccessible(SCHEMA_NAME, PROD_CLUSTER);
131+
StepVerifier.create(consumerGroupAccessibleMono)
132+
.expectNext(true)
133+
.expectComplete()
134+
.verify();
135+
});
136+
}
137+
138+
@Test
139+
void isConnectAccessible() {
140+
withSecurityContext(() -> {
141+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
142+
Mono<Boolean> consumerGroupAccessibleMono =
143+
accessControlService.isConnectAccessible(CONNECT_NAME, PROD_CLUSTER);
144+
StepVerifier.create(consumerGroupAccessibleMono)
145+
.expectNext(true)
146+
.expectComplete()
147+
.verify();
148+
});
149+
}
150+
151+
@Test
152+
void isConnectAccessibleDto() {
153+
withSecurityContext(() -> {
154+
when(user.groups()).thenReturn(List.of(DEV_ROLE));
155+
ConnectDTO connectDto = ConnectDTO.builder()
156+
.name(CONNECT_NAME)
157+
.build();
158+
Mono<Boolean> consumerGroupAccessibleMono =
159+
accessControlService.isConnectAccessible(connectDto, PROD_CLUSTER);
160+
StepVerifier.create(consumerGroupAccessibleMono)
161+
.expectNext(true)
162+
.expectComplete()
163+
.verify();
164+
});
165+
}
166+
167+
@Test
168+
void getRoles() {
169+
List<Role> roles = accessControlService.getRoles();
170+
assertThat(roles).isEmpty();
171+
}
172+
173+
}

0 commit comments

Comments
 (0)