scriptoriumadmin/skillhub
1
0
Fork 0
mirror of https://github.com/iflytek/skillhub.git synced 2026-08-28 11:25:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix: migrate scanner stream to redisson and refine audit status labels` (#153)

Browse source 
* fix(scanner): migrate scan stream to redisson

* fix(web): update security audit pending labels
This commit is contained in:
XiaoSeS 2026-03-24 14:56:16 +08:00 • committed by GitHub
parent eda8764b30
commit 747f649fb7
23 changed files with 592 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
server/skillhub-app/pom.xml
View file

@ -77,6 +77,11 @@
<groupId>org.springframework.session</groupId>
<artifactId>spring-session-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.51.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>

32
server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedisStreamConfig.java
View file

@ -6,14 +6,14 @@ import com.iflytek.skillhub.domain.security.SecurityScanService;
import com.iflytek.skillhub.domain.security.SecurityScanner;
import com.iflytek.skillhub.domain.skill.SkillRepository;
import com.iflytek.skillhub.domain.skill.SkillVersionRepository;
import com.iflytek.skillhub.stream.RedisScanTaskProducer;
import com.iflytek.skillhub.stream.RedissonScanTaskProducer;
import com.iflytek.skillhub.stream.ScanTaskConsumer;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.time.Duration;
@Configuration
@ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true")
@ -25,13 +25,25 @@ public class RedisStreamConfig {
@Value("${skillhub.security.stream.group:skillhub-scanners}")
private String groupName;
@Value("${skillhub.security.stream.reclaim-enabled:true}")
private boolean reclaimEnabled;
@Value("${skillhub.security.stream.reclaim-min-idle:PT2M}")
private Duration reclaimMinIdle;
@Value("${skillhub.security.stream.reclaim-batch-size:20}")
private int reclaimBatchSize;
@Value("${skillhub.security.stream.reclaim-interval:PT30S}")
private Duration reclaimInterval;
@Bean
public RedisScanTaskProducer redisScanTaskProducer(StringRedisTemplate redisTemplate) {
return new RedisScanTaskProducer(redisTemplate, streamKey);
public RedissonScanTaskProducer redisScanTaskProducer(RedissonClient redissonClient) {
return new RedissonScanTaskProducer(redissonClient, streamKey);
}
@Bean
public ScanTaskConsumer scanTaskConsumer(RedisConnectionFactory connectionFactory,
public ScanTaskConsumer scanTaskConsumer(RedissonClient redissonClient,
SecurityScanner securityScanner,
SecurityScanService securityScanService,
SkillVersionRepository skillVersionRepository,
@ -39,7 +51,7 @@ public class RedisStreamConfig {
ReviewTaskRepository reviewTaskRepository,
ScanTaskProducer scanTaskProducer) {
return new ScanTaskConsumer(
connectionFactory,
redissonClient,
streamKey,
groupName,
securityScanner,
@ -47,7 +59,11 @@ public class RedisStreamConfig {
skillVersionRepository,
skillRepository,
reviewTaskRepository,
scanTaskProducer
scanTaskProducer,
reclaimEnabled,
reclaimMinIdle,
reclaimBatchSize,
reclaimInterval
);
}
}

56
server/skillhub-app/src/main/java/com/iflytek/skillhub/config/RedissonConfig.java Normal file
View file

@ -0,0 +1,56 @@
package com.iflytek.skillhub.config;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;
@Configuration
@ConditionalOnProperty(prefix = "skillhub.security.scanner", name = "enabled", havingValue = "true")
public class RedissonConfig {
@Bean(destroyMethod = "shutdown")
public RedissonClient redissonClient(RedisProperties redisProperties) {
return Redisson.create(createConfig(redisProperties));
}
static Config createConfig(RedisProperties redisProperties) {
Config config = new Config();
SingleServerConfig singleServerConfig = config.useSingleServer()
.setAddress(resolveAddress(redisProperties))
.setDatabase(redisProperties.getDatabase());
if (StringUtils.hasText(redisProperties.getUsername())) {
singleServerConfig.setUsername(redisProperties.getUsername());
}
if (StringUtils.hasText(redisProperties.getPassword())) {
singleServerConfig.setPassword(redisProperties.getPassword());
}
if (StringUtils.hasText(redisProperties.getClientName())) {
singleServerConfig.setClientName(redisProperties.getClientName());
}
if (redisProperties.getTimeout() != null) {
singleServerConfig.setTimeout(Math.toIntExact(redisProperties.getTimeout().toMillis()));
}
if (redisProperties.getConnectTimeout() != null) {
singleServerConfig.setConnectTimeout(Math.toIntExact(redisProperties.getConnectTimeout().toMillis()));
}
return config;
}
private static String resolveAddress(RedisProperties redisProperties) {
if (StringUtils.hasText(redisProperties.getUrl())) {
return redisProperties.getUrl();
}
String scheme = redisProperties.getSsl() != null && redisProperties.getSsl().isEnabled()
? "rediss"
: "redis";
return scheme + "://" + redisProperties.getHost() + ":" + redisProperties.getPort();
}
}

229
server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/AbstractStreamConsumer.java
View file

@ -2,113 +2,214 @@ package com.iflytek.skillhub.stream;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.redisson.api.AutoClaimResult;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamCreateGroupArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.redisson.client.codec.StringCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import java.time.Duration;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class AbstractStreamConsumer<T> implements StreamListener<String, MapRecord<String, String, String>> {
public abstract class AbstractStreamConsumer<T> {
protected final Logger log = LoggerFactory.getLogger(getClass());
private static final String FIELD_RETRY_COUNT = "retryCount";
private static final int MAX_RETRY_COUNT = 3;
private static final int READ_BATCH_SIZE = 10;
private static final Duration POLL_TIMEOUT = Duration.ofSeconds(2);
private final RedisConnectionFactory connectionFactory;
private final RedissonClient redissonClient;
private final String streamKey;
private final String groupName;
private final String consumerName;
private StringRedisTemplate redisTemplate;
private final boolean reclaimEnabled;
private final Duration reclaimMinIdle;
private final int reclaimBatchSize;
private final Duration reclaimInterval;
private final AtomicBoolean running = new AtomicBoolean(false);
private StreamMessageListenerContainer<String, MapRecord<String, String, String>> container;
private RStream<String, String> stream;
private ExecutorService consumerExecutor;
private ScheduledExecutorService reclaimExecutor;
protected AbstractStreamConsumer(RedisConnectionFactory connectionFactory,
protected AbstractStreamConsumer(RedissonClient redissonClient,
String streamKey,
String groupName) {
this.connectionFactory = connectionFactory;
this(redissonClient, streamKey, groupName, true, Duration.ofMinutes(2), 20, Duration.ofSeconds(30));
}
protected AbstractStreamConsumer(RedissonClient redissonClient,
String streamKey,
String groupName,
boolean reclaimEnabled,
Duration reclaimMinIdle,
int reclaimBatchSize,
Duration reclaimInterval) {
this.redissonClient = redissonClient;
this.streamKey = streamKey;
this.groupName = groupName;
this.reclaimEnabled = reclaimEnabled;
this.reclaimMinIdle = reclaimMinIdle;
this.reclaimBatchSize = reclaimBatchSize;
this.reclaimInterval = reclaimInterval;
this.consumerName = consumerPrefix() + "-" + UUID.randomUUID().toString().substring(0, 8);
}
@PostConstruct
public void init() {
if (connectionFactory == null) {
if (redissonClient == null) {
return;
}
this.redisTemplate = createRedisTemplate();
this.stream = createStream();
initializeStreamAndGroup();
startConsumer();
startPendingReclaimer();
}
@PreDestroy
public void shutdown() {
if (container != null) {
container.stop();
running.set(false);
if (consumerExecutor != null) {
consumerExecutor.shutdownNow();
}
if (reclaimExecutor != null) {
reclaimExecutor.shutdownNow();
}
}
private void initializeStreamAndGroup() {
try {
StringRedisTemplate template = redisTemplate();
if (Boolean.FALSE.equals(template.hasKey(streamKey))) {
template.opsForStream().add(streamKey, Map.of("_init", "true"));
}
try {
template.opsForStream().createGroup(streamKey, ReadOffset.from("0"), groupName);
} catch (Exception e) {
if (e.getMessage() == null || !e.getMessage().contains("BUSYGROUP")) {
log.warn("Failed to create consumer group: stream={}, group={}", streamKey, groupName, e);
}
}
stream().createGroup(StreamCreateGroupArgs.name(groupName).makeStream());
} catch (Exception e) {
throw new RuntimeException("Failed to initialize Redis Stream consumer", e);
if (!isConsumerGroupAlreadyExists(e)) {
log.warn("Failed to create consumer group: stream={}, group={}", streamKey, groupName, e);
}
}
}
static boolean isConsumerGroupAlreadyExists(Throwable error) {
Throwable current = error;
while (current != null) {
if (current.getMessage() != null && current.getMessage().contains("BUSYGROUP")) {
return true;
}
current = current.getCause();
}
return false;
}
private void startConsumer() {
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
.pollTimeout(Duration.ofSeconds(2))
.build();
container = StreamMessageListenerContainer.create(connectionFactory, options);
Subscription ignored = container.receive(
Consumer.from(groupName, consumerName),
StreamOffset.create(streamKey, ReadOffset.lastConsumed()),
this
);
container.start();
running.set(true);
consumerExecutor = Executors.newSingleThreadExecutor(threadFactory(consumerPrefix() + "-consumer"));
consumerExecutor.submit(this::consumeLoop);
}
@Override
public void onMessage(MapRecord<String, String, String> message) {
T payload = parsePayload(message.getId().getValue(), message.getValue());
private void startPendingReclaimer() {
if (!reclaimEnabled) {
return;
}
reclaimExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory(consumerPrefix() + "-reclaimer"));
long intervalMillis = Math.max(1L, reclaimInterval.toMillis());
reclaimExecutor.scheduleWithFixedDelay(this::safeReclaimPendingMessages,
intervalMillis, intervalMillis, TimeUnit.MILLISECONDS);
}
private void consumeLoop() {
while (running.get()) {
try {
consumeAvailableMessages();
} catch (Exception e) {
if (Thread.currentThread().isInterrupted()) {
Thread.currentThread().interrupt();
break;
}
log.error("Failed while consuming stream messages: stream={}, group={}, consumer={}",
streamKey, groupName, consumerName, e);
}
}
}
private void safeReclaimPendingMessages() {
try {
reclaimPendingMessages();
} catch (Exception e) {
log.error("Failed while reclaiming pending stream messages: stream={}, group={}, consumer={}",
streamKey, groupName, consumerName, e);
}
}
void consumeAvailableMessages() {
Map<StreamMessageId, Map<String, String>> messages = stream().readGroup(
groupName,
consumerName,
StreamReadGroupArgs.neverDelivered()
.count(READ_BATCH_SIZE)
.timeout(POLL_TIMEOUT)
);
processMessages(messages);
}
void reclaimPendingMessages() {
if (!reclaimEnabled) {
return;
}
StreamMessageId startId = StreamMessageId.MIN;
while (true) {
AutoClaimResult<String, String> result = stream().autoClaim(
groupName,
consumerName,
reclaimMinIdle.toMillis(),
TimeUnit.MILLISECONDS,
startId,
reclaimBatchSize
);
if (result == null || result.getMessages() == null || result.getMessages().isEmpty()) {
return;
}
processMessages(result.getMessages());
if (result.getMessages().size() < reclaimBatchSize || result.getNextId() == null) {
return;
}
startId = result.getNextId();
}
}
private void processMessages(Map<StreamMessageId, Map<String, String>> messages) {
if (messages == null || messages.isEmpty()) {
return;
}
messages.forEach(this::handleMessage);
}
void handleMessage(StreamMessageId messageId, Map<String, String> data) {
T payload = parsePayload(messageId.toString(), data);
if (payload == null) {
acknowledge(message);
acknowledge(messageId);
return;
}
int retryCount = parseRetryCount(message.getValue());
int retryCount = parseRetryCount(data);
try {
markProcessing(payload);
processBusiness(payload);
markCompleted(payload);
acknowledge(message);
acknowledge(messageId);
} catch (Exception e) {
handleFailure(payload, retryCount, e);
acknowledge(message);
acknowledge(messageId);
}
}
@ -137,19 +238,31 @@ public abstract class AbstractStreamConsumer<T> implements StreamListener<String
return error.length() > 500 ? error.substring(0, 500) : error;
}
protected StringRedisTemplate createRedisTemplate() {
return new StringRedisTemplate(connectionFactory);
protected RStream<String, String> createStream() {
return redissonClient.getStream(streamKey, StringCodec.INSTANCE);
}
protected void acknowledge(MapRecord<String, String, String> message) {
redisTemplate().opsForStream().acknowledge(streamKey, groupName, message.getId());
protected void acknowledge(StreamMessageId messageId) {
stream().ack(groupName, messageId);
}
private StringRedisTemplate redisTemplate() {
if (redisTemplate == null) {
redisTemplate = createRedisTemplate();
protected final RStream<String, String> stream() {
if (stream == null) {
stream = createStream();
}
return redisTemplate;
return stream;
}
protected final String consumerName() {
return consumerName;
}
private ThreadFactory threadFactory(String namePrefix) {
return runnable -> {
Thread thread = new Thread(runnable, namePrefix + "-" + consumerName);
thread.setDaemon(true);
return thread;
};
}
protected abstract String taskDisplayName();

25
server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedisScanTaskProducer.java → server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/RedissonScanTaskProducer.java
View file

@ -2,25 +2,26 @@ package com.iflytek.skillhub.stream;
import com.iflytek.skillhub.domain.security.ScanTask;
import com.iflytek.skillhub.domain.security.ScanTaskProducer;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.client.codec.StringCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.connection.stream.StringRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.HashMap;
import java.util.Map;
public class RedisScanTaskProducer implements ScanTaskProducer {
public class RedissonScanTaskProducer implements ScanTaskProducer {
private static final Logger log = LoggerFactory.getLogger(RedisScanTaskProducer.class);
private static final Logger log = LoggerFactory.getLogger(RedissonScanTaskProducer.class);
private final StringRedisTemplate redisTemplate;
private final RedissonClient redissonClient;
private final String streamKey;
public RedisScanTaskProducer(StringRedisTemplate redisTemplate, String streamKey) {
this.redisTemplate = redisTemplate;
public RedissonScanTaskProducer(RedissonClient redissonClient, String streamKey) {
this.redissonClient = redissonClient;
this.streamKey = streamKey;
}
@ -36,9 +37,9 @@ public class RedisScanTaskProducer implements ScanTaskProducer {
fields.putAll(task.metadata());
}
StringRecord record = StreamRecords.string(fields).withStreamKey(streamKey);
RecordId recordId = redisTemplate.opsForStream().add(record);
RStream<String, String> stream = redissonClient.getStream(streamKey, StringCodec.INSTANCE);
StreamMessageId messageId = stream.add(StreamAddArgs.entries(fields));
log.info("Published scan task: taskId={}, versionId={}, recordId={}",
task.taskId(), task.versionId(), recordId);
task.taskId(), task.versionId(), messageId);
}
}

29
server/skillhub-app/src/main/java/com/iflytek/skillhub/stream/ScanTaskConsumer.java
View file

@ -12,12 +12,13 @@ import com.iflytek.skillhub.domain.security.SecurityScanner;
import com.iflytek.skillhub.domain.skill.SkillRepository;
import com.iflytek.skillhub.domain.skill.SkillVersionRepository;
import com.iflytek.skillhub.domain.skill.SkillVersionStatus;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.redisson.api.RedissonClient;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Comparator;
import java.util.Map;
@ -31,7 +32,7 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
private final ReviewTaskRepository reviewTaskRepository;
private final ScanTaskProducer scanTaskProducer;
public ScanTaskConsumer(RedisConnectionFactory connectionFactory,
public ScanTaskConsumer(RedissonClient redissonClient,
String streamKey,
String groupName,
SecurityScanner securityScanner,
@ -40,7 +41,29 @@ public class ScanTaskConsumer extends AbstractStreamConsumer<ScanTaskConsumer.Sc
SkillRepository skillRepository,
ReviewTaskRepository reviewTaskRepository,
ScanTaskProducer scanTaskProducer) {
super(connectionFactory, streamKey, groupName);
super(redissonClient, streamKey, groupName);
this.securityScanner = securityScanner;
this.securityScanService = securityScanService;
this.skillVersionRepository = skillVersionRepository;
this.skillRepository = skillRepository;
this.reviewTaskRepository = reviewTaskRepository;
this.scanTaskProducer = scanTaskProducer;
}
public ScanTaskConsumer(RedissonClient redissonClient,
String streamKey,
String groupName,
SecurityScanner securityScanner,
SecurityScanService securityScanService,
SkillVersionRepository skillVersionRepository,
SkillRepository skillRepository,
ReviewTaskRepository reviewTaskRepository,
ScanTaskProducer scanTaskProducer,
boolean reclaimEnabled,
Duration reclaimMinIdle,
int reclaimBatchSize,
Duration reclaimInterval) {
super(redissonClient, streamKey, groupName, reclaimEnabled, reclaimMinIdle, reclaimBatchSize, reclaimInterval);
this.securityScanner = securityScanner;
this.securityScanService = securityScanService;
this.skillVersionRepository = skillVersionRepository;

5
server/skillhub-app/src/main/resources/application-local.yml
View file

@ -34,6 +34,11 @@ skillhub:
security:
scanner:
enabled: ${SKILLHUB_SECURITY_SCANNER_ENABLED:false}
stream:
reclaim-enabled: ${SKILLHUB_SCAN_STREAM_RECLAIM_ENABLED:true}
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT2M}
reclaim-batch-size: ${SKILLHUB_SCAN_STREAM_RECLAIM_BATCH_SIZE:20}
reclaim-interval: ${SKILLHUB_SCAN_STREAM_RECLAIM_INTERVAL:PT30S}
bootstrap:
admin:
enabled: ${BOOTSTRAP_ADMIN_ENABLED:true}

4
server/skillhub-app/src/main/resources/application.yml
View file

@ -153,6 +153,10 @@ skillhub:
stream:
key: ${SKILLHUB_SCAN_STREAM_KEY:skillhub:scan:requests}
group: ${SKILLHUB_SCAN_STREAM_GROUP:skillhub-scanners}
reclaim-enabled: ${SKILLHUB_SCAN_STREAM_RECLAIM_ENABLED:true}
reclaim-min-idle: ${SKILLHUB_SCAN_STREAM_RECLAIM_MIN_IDLE:PT2M}
reclaim-batch-size: ${SKILLHUB_SCAN_STREAM_RECLAIM_BATCH_SIZE:20}
reclaim-interval: ${SKILLHUB_SCAN_STREAM_RECLAIM_INTERVAL:PT30S}
bootstrap:
admin:
enabled: ${BOOTSTRAP_ADMIN_ENABLED:false}

62
server/skillhub-app/src/test/java/com/iflytek/skillhub/config/RedissonConfigTest.java Normal file
View file

@ -0,0 +1,62 @@
package com.iflytek.skillhub.config;
import org.junit.jupiter.api.Test;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import java.time.Duration;
import static org.assertj.core.api.Assertions.assertThat;
class RedissonConfigTest {
@Test
void createConfig_buildsRedisAddressFromHostPortAndSslFlag() {
RedisProperties properties = new RedisProperties();
properties.setHost("redis.internal");
properties.setPort(6380);
properties.getSsl().setEnabled(true);
Config config = RedissonConfig.createConfig(properties);
SingleServerConfig serverConfig = config.useSingleServer();
assertThat(serverConfig.getAddress()).isEqualTo("rediss://redis.internal:6380");
}
@Test
void createConfig_prefersExplicitRedisUrlWhenPresent() {
RedisProperties properties = new RedisProperties();
properties.setUrl("redis://cache.example:6379");
properties.setHost("ignored-host");
properties.setPort(6380);
Config config = RedissonConfig.createConfig(properties);
SingleServerConfig serverConfig = config.useSingleServer();
assertThat(serverConfig.getAddress()).isEqualTo("redis://cache.example:6379");
}
@Test
void createConfig_appliesDatabaseCredentialsClientNameAndTimeouts() {
RedisProperties properties = new RedisProperties();
properties.setHost("localhost");
properties.setPort(6379);
properties.setDatabase(5);
properties.setUsername("skillhub");
properties.setPassword("secret");
properties.setClientName("skillhub-stream");
properties.setTimeout(Duration.ofSeconds(3));
properties.setConnectTimeout(Duration.ofSeconds(5));
Config config = RedissonConfig.createConfig(properties);
SingleServerConfig serverConfig = config.useSingleServer();
assertThat(serverConfig.getDatabase()).isEqualTo(5);
assertThat(serverConfig.getUsername()).isEqualTo("skillhub");
assertThat(serverConfig.getPassword()).isEqualTo("secret");
assertThat(serverConfig.getClientName()).isEqualTo("skillhub-stream");
assertThat(serverConfig.getTimeout()).isEqualTo(3000);
assertThat(serverConfig.getConnectTimeout()).isEqualTo(5000);
}
}

165
server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/AbstractStreamConsumerTest.java
View file

@ -1,88 +1,131 @@
package com.iflytek.skillhub.stream;
import io.lettuce.core.RedisBusyException;
import org.junit.jupiter.api.Test;
import org.redisson.api.AutoClaimResult;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.springframework.data.redis.RedisSystemException;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import static org.mockito.Mockito.when;
class AbstractStreamConsumerTest {
@Test
void onMessage_acknowledgesAfterSuccessfulProcessing() {
StreamOperations<String, Object, Object> streamOperations = mock(StreamOperations.class);
StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class);
org.mockito.Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations);
TestConsumer consumer = new TestConsumer(redisTemplate);
MapRecord<String, String, String> message = StreamRecords.newRecord()
.in("scan-stream")
.withId(RecordId.of("1-0"))
.ofMap(Map.of("payload", "ok"));
void handleMessage_acknowledgesAfterSuccessfulProcessing() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
TestConsumer consumer = new TestConsumer(stream);
StreamMessageId messageId = new StreamMessageId(1, 0);
consumer.onMessage(message);
consumer.handleMessage(messageId, Map.of("payload", "ok"));
verify(streamOperations).acknowledge("scan-stream", "scan-group", message.getId());
verify(stream).ack("scan-group", messageId);
}
@Test
void onMessage_acknowledgesAfterRetryableFailure() {
StreamOperations<String, Object, Object> streamOperations = mock(StreamOperations.class);
StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class);
org.mockito.Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations);
TestConsumer consumer = new TestConsumer(redisTemplate);
void handleMessage_acknowledgesAfterRetryableFailure() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
TestConsumer consumer = new TestConsumer(stream);
consumer.fail = true;
MapRecord<String, String, String> message = StreamRecords.newRecord()
.in("scan-stream")
.withId(RecordId.of("2-0"))
.ofMap(Map.of("payload", "boom"));
StreamMessageId messageId = new StreamMessageId(2, 0);
consumer.onMessage(message);
consumer.handleMessage(messageId, Map.of("payload", "boom"));
verify(streamOperations).acknowledge("scan-stream", "scan-group", message.getId());
verify(streamOperations, times(1)).acknowledge("scan-stream", "scan-group", message.getId());
verify(stream).ack("scan-group", messageId);
verify(stream, times(1)).ack("scan-group", messageId);
}
@Test
void onMessage_reusesRedisTemplateForAcknowledgement() {
StreamOperations<String, Object, Object> streamOperations = mock(StreamOperations.class);
StringRedisTemplate redisTemplate = mock(StringRedisTemplate.class);
org.mockito.Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations);
CountingConsumer consumer = new CountingConsumer(redisTemplate);
MapRecord<String, String, String> first = StreamRecords.newRecord()
.in("scan-stream")
.withId(RecordId.of("3-0"))
.ofMap(Map.of("payload", "one"));
MapRecord<String, String, String> second = StreamRecords.newRecord()
.in("scan-stream")
.withId(RecordId.of("4-0"))
.ofMap(Map.of("payload", "two"));
void consumeAvailableMessages_processesNeverDeliveredMessages() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
TestConsumer consumer = new TestConsumer(stream);
StreamMessageId first = new StreamMessageId(3, 0);
StreamMessageId second = new StreamMessageId(4, 0);
Map<StreamMessageId, Map<String, String>> messages = new LinkedHashMap<>();
messages.put(first, Map.of("payload", "one"));
messages.put(second, Map.of("payload", "two"));
when(stream.readGroup(eq("scan-group"), anyString(), org.mockito.ArgumentMatchers.<StreamReadGroupArgs>any()))
.thenReturn(messages);
consumer.onMessage(first);
consumer.onMessage(second);
consumer.consumeAvailableMessages();
org.junit.jupiter.api.Assertions.assertEquals(1, consumer.templateCreationCount.get());
verify(stream).ack("scan-group", first);
verify(stream).ack("scan-group", second);
}
@Test
void reclaimPendingMessages_autoClaimsAndProcessesMessages() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
TestConsumer consumer = new TestConsumer(stream);
StreamMessageId reclaimedId = new StreamMessageId(5, 0);
when(stream.autoClaim(eq("scan-group"), anyString(), anyLong(), eq(java.util.concurrent.TimeUnit.MILLISECONDS),
eq(StreamMessageId.MIN), anyInt()))
.thenReturn(new AutoClaimResult<>(
StreamMessageId.MAX,
Map.of(reclaimedId, Map.of("payload", "reclaimed")),
java.util.List.of()
));
consumer.reclaimPendingMessages();
verify(stream).ack("scan-group", reclaimedId);
}
@Test
void handleMessage_reusesStreamInstanceForAcknowledgement() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
CountingConsumer consumer = new CountingConsumer(stream);
consumer.handleMessage(new StreamMessageId(6, 0), Map.of("payload", "one"));
consumer.handleMessage(new StreamMessageId(7, 0), Map.of("payload", "two"));
assertThat(consumer.streamCreationCount.get()).isEqualTo(1);
}
@Test
void detectsBusyGroupWhenWrappedInRedisSystemException() {
RedisSystemException wrapped = new RedisSystemException(
"Error in execution",
new RedisBusyException("BUSYGROUP Consumer Group name already exists")
);
assertThat(AbstractStreamConsumer.isConsumerGroupAlreadyExists(wrapped)).isTrue();
}
private static class TestConsumer extends AbstractStreamConsumer<String> {
private final StringRedisTemplate redisTemplate;
private final RStream<String, String> stream;
private boolean fail;
private TestConsumer(StringRedisTemplate redisTemplate) {
super(mock(RedisConnectionFactory.class), "scan-stream", "scan-group");
this.redisTemplate = redisTemplate;
private TestConsumer(RStream<String, String> stream) {
super(mock(RedissonClient.class), "scan-stream", "scan-group", true, Duration.ofMinutes(2), 20, Duration.ofSeconds(30));
this.stream = stream;
}
@Override
protected StringRedisTemplate createRedisTemplate() {
return redisTemplate;
protected RStream<String, String> createStream() {
return stream;
}
@Override
@ -130,16 +173,16 @@ class AbstractStreamConsumerTest {
}
private static final class CountingConsumer extends TestConsumer {
private final AtomicInteger templateCreationCount = new AtomicInteger();
private final AtomicInteger streamCreationCount = new AtomicInteger();
private CountingConsumer(StringRedisTemplate redisTemplate) {
super(redisTemplate);
private CountingConsumer(RStream<String, String> stream) {
super(stream);
}
@Override
protected StringRedisTemplate createRedisTemplate() {
templateCreationCount.incrementAndGet();
return super.createRedisTemplate();
protected RStream<String, String> createStream() {
streamCreationCount.incrementAndGet();
return super.createStream();
}
}
}

48
server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/RedissonScanTaskProducerTest.java Normal file
View file

@ -0,0 +1,48 @@
package com.iflytek.skillhub.stream;
import com.iflytek.skillhub.domain.security.ScanTask;
import org.junit.jupiter.api.Test;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.client.codec.StringCodec;
import org.redisson.api.stream.StreamAddArgs;
import org.mockito.ArgumentCaptor;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class RedissonScanTaskProducerTest {
@Test
void publishScanTask_writesExpectedFieldsToConfiguredStream() {
@SuppressWarnings("unchecked")
RStream<String, String> stream = mock(RStream.class);
@SuppressWarnings("unchecked")
RStream<String, String> typedStream = (RStream<String, String>) (RStream<?, ?>) stream;
RedissonClient redissonClient = mock(RedissonClient.class);
doReturn(typedStream).when(redissonClient).getStream("skillhub:scan:requests", StringCodec.INSTANCE);
when(stream.add(any())).thenReturn(new StreamMessageId(1, 0));
RedissonScanTaskProducer producer = new RedissonScanTaskProducer(redissonClient, "skillhub:scan:requests");
producer.publishScanTask(new ScanTask(
"task-1",
42L,
"/tmp/skill",
"publisher-1",
1711260000000L,
Map.of("scannerType", "skill-scanner")
));
verify(redissonClient).getStream("skillhub:scan:requests", StringCodec.INSTANCE);
ArgumentCaptor<StreamAddArgs<String, String>> argsCaptor = ArgumentCaptor.forClass(StreamAddArgs.class);
verify(stream).add(argsCaptor.capture());
assertThat(argsCaptor.getValue()).isNotNull();
}
}

4
server/skillhub-app/src/test/java/com/iflytek/skillhub/stream/ScanTaskConsumerPathSafetyTest.java
View file

@ -12,14 +12,14 @@ import java.lang.reflect.Method;
import java.nio.file.Files;
import java.nio.file.Path;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.redisson.api.RedissonClient;
class ScanTaskConsumerPathSafetyTest {
@Test
void cleanupTempPath_ignoresPathsOutsideScanTempDirectory() throws Exception {
ScanTaskConsumer consumer = new ScanTaskConsumer(
org.mockito.Mockito.mock(RedisConnectionFactory.class),
org.mockito.Mockito.mock(RedissonClient.class),
"scan-stream",
"scan-group",
org.mockito.Mockito.mock(SecurityScanner.class),

7
web/src/features/security-audit/display-state.ts Normal file
View file

@ -0,0 +1,7 @@
import type { SecurityAuditDisplayState, SecurityAuditRecord } from './types'
export function getSecurityAuditDisplayState(
audit: Pick<SecurityAuditRecord, 'scannedAt' | 'verdict'>
): SecurityAuditDisplayState {
return audit.scannedAt ? audit.verdict : 'SCANNING'
}

10
web/src/features/security-audit/security-audit-section.test.tsx
View file

@ -98,6 +98,16 @@ describe('SecurityAuditSection', () => {
expect(html).toContain('securityAudit.verdict.SUSPICIOUS')
})
it('renders scanning status when the audit has not completed yet', () => {
mockAudits = [createAudit({ verdict: 'SUSPICIOUS', scannedAt: null })]
mockIsLoading = false
const html = renderToStaticMarkup(<SecurityAuditSection skillId={1} versionId={10} />)
expect(html).toContain('securityAudit.statusScanning')
expect(html).not.toContain('securityAudit.verdict.SUSPICIOUS')
})
it('renders the findings count', () => {
mockAudits = [createAudit({ findingsCount: 3 })]
mockIsLoading = false

3
web/src/features/security-audit/security-audit-section.tsx
View file

@ -4,6 +4,7 @@ import { ChevronDown, ChevronUp, Shield } from 'lucide-react'
import { Card } from '@/shared/ui/card'
import { Button } from '@/shared/ui/button'
import { useSecurityAudits } from './use-security-audit'
import { getSecurityAuditDisplayState } from './display-state'
import { VerdictBadge } from './verdict-badge'
import { FindingItem } from './finding-item'
import type { FindingSeverity, SecurityAuditRecord } from './types'
@ -72,7 +73,7 @@ function ScannerCard({ audit }: { audit: SecurityAuditRecord }) {
<div className="flex items-center justify-between flex-wrap gap-2">
<div className="flex items-center gap-3">
<span className="text-sm font-semibold font-mono">{audit.scannerType}</span>
<VerdictBadge verdict={audit.verdict} />
<VerdictBadge displayState={getSecurityAuditDisplayState(audit)} />
</div>
<div className="flex items-center gap-4 text-sm text-muted-foreground">
<span>

9
web/src/features/security-audit/security-audit-summary.test.tsx
View file

@ -97,6 +97,15 @@ describe('SecurityAuditSummary', () => {
expect(html).toContain('securityAudit.verdict.BLOCKED')
})
it('renders scanning status when the audit has not completed yet', () => {
mockAudits = [createAudit({ verdict: 'SUSPICIOUS', scannedAt: null })]
const html = renderToStaticMarkup(<SecurityAuditSummary skillId={1} versionId={10} />)
expect(html).toContain('securityAudit.statusScanning')
expect(html).not.toContain('securityAudit.verdict.SUSPICIOUS')
})
it('renders the total findings count across all audits', () => {
mockAudits = [
createAudit({ id: 1, findingsCount: 3 }),

3
web/src/features/security-audit/security-audit-summary.tsx
View file

@ -5,6 +5,7 @@ import { Card } from '@/shared/ui/card'
import { Button } from '@/shared/ui/button'
import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogDescription } from '@/shared/ui/dialog'
import { useSecurityAudits } from './use-security-audit'
import { getSecurityAuditDisplayState } from './display-state'
import { VerdictBadge } from './verdict-badge'
import { SecurityAuditSection } from './security-audit-section'
@ -40,7 +41,7 @@ export function SecurityAuditSummary({ skillId, versionId }: SecurityAuditSummar
className="flex items-center justify-between rounded-xl border border-border/60 bg-secondary/20 p-3"
>
<span className="text-xs font-mono text-muted-foreground">{audit.scannerType}</span>
<VerdictBadge verdict={audit.verdict} />
<VerdictBadge displayState={getSecurityAuditDisplayState(audit)} />
</div>
))}
</div>

1
web/src/features/security-audit/types.ts
View file

@ -1,4 +1,5 @@
export type SecurityVerdict = 'SAFE' | 'SUSPICIOUS' | 'DANGEROUS' | 'BLOCKED'
export type SecurityAuditDisplayState = SecurityVerdict | 'SCANNING'
export type FindingSeverity = 'CRITICAL' | 'HIGH' | 'MEDIUM' | 'LOW' | 'INFO'
export interface SecurityFinding {

6
web/src/features/security-audit/verdict-badge.test.tsx
View file

@ -47,6 +47,12 @@ describe('VerdictBadge', () => {
expect(html).toContain('text-red-700')
})
it('renders the scanning label for pending scan display state', () => {
const html = renderToStaticMarkup(<VerdictBadge displayState="SCANNING" />)
expect(html).toContain('securityAudit.statusScanning')
})
it('renders as a span with rounded-full pill styling', () => {
const html = renderToStaticMarkup(<VerdictBadge verdict="SAFE" />)

21
web/src/features/security-audit/verdict-badge.tsx
View file

@ -1,25 +1,36 @@
import { useTranslation } from 'react-i18next'
import type { SecurityVerdict } from './types'
import type { SecurityAuditDisplayState, SecurityVerdict } from './types'
interface VerdictBadgeProps {
verdict: SecurityVerdict
verdict?: SecurityVerdict
displayState?: SecurityAuditDisplayState
}
export function VerdictBadge({ verdict }: VerdictBadgeProps) {
export function VerdictBadge({ verdict, displayState }: VerdictBadgeProps) {
const { t } = useTranslation()
const state = displayState ?? verdict
if (!state) {
return null
}
const styles = {
SCANNING: 'bg-sky-500/10 text-sky-700 dark:text-sky-400',
SAFE: 'bg-emerald-500/10 text-emerald-700 dark:text-emerald-400',
SUSPICIOUS: 'bg-amber-500/10 text-amber-700 dark:text-amber-400',
DANGEROUS: 'bg-orange-500/10 text-orange-700 dark:text-orange-400',
BLOCKED: 'bg-red-500/10 text-red-700 dark:text-red-400',
}
const label = state === 'SCANNING'
? t('securityAudit.statusScanning')
: t(`securityAudit.verdict.${state}`)
return (
<span
className={`rounded-full px-2.5 py-0.5 text-sm font-medium ${styles[verdict]}`}
className={`rounded-full px-2.5 py-0.5 text-sm font-medium ${styles[state]}`}
>
{t(`securityAudit.verdict.${verdict}`)}
{label}
</span>
)
}

3
web/src/i18n/locales/en.json
View file

@ -1268,13 +1268,14 @@
"findingsCount": "{{count}} findings",
"totalFindings": "{{count}} findings total",
"scanDuration": "{{seconds}}s",
"statusScanning": "Scanning",
"remediation": "Remediation",
"viewDetails": "View Details",
"verdict": {
"SAFE": "Safe",
"SUSPICIOUS": "Suspicious",
"DANGEROUS": "Dangerous",
"BLOCKED": "Blocked"
"BLOCKED": "High Risk"
},
"severity": {
"CRITICAL": "Critical",

3
web/src/i18n/locales/zh.json
View file

@ -1268,13 +1268,14 @@
"findingsCount": "{{count}} 项发现",
"totalFindings": "共 {{count}} 项发现",
"scanDuration": "{{seconds}}s",
"statusScanning": "扫描中",
"remediation": "修复建议",
"viewDetails": "查看详情",
"verdict": {
"SAFE": "安全",
"SUSPICIOUS": "可疑",
"DANGEROUS": "危险",
"BLOCKED": "已拦截"
"BLOCKED": "高风险"
},
"severity": {
"CRITICAL": "严重",

15
web/src/i18n/security-audit-locale.test.ts Normal file
View file

@ -0,0 +1,15 @@
import { describe, expect, it } from 'vitest'
import en from './locales/en.json'
import zh from './locales/zh.json'
describe('security audit locales', () => {
it('defines the scanning label in both locales', () => {
expect(zh.securityAudit.statusScanning).toBe('扫描中')
expect(en.securityAudit.statusScanning).toBe('Scanning')
})
it('uses the updated blocked wording in both locales', () => {
expect(zh.securityAudit.verdict.BLOCKED).toBe('高风险')
expect(en.securityAudit.verdict.BLOCKED).toBe('High Risk')
})
})