diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java index 1039309..df1182e 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java @@ -63,8 +63,30 @@ public class NormalizationCacheInvalidationService { } } + /** + * 清空规则专属 Spring Cache，并按设备精确广播 JetCache BOTH 失效。 + * + *

JetCache 没有安全的全库 clear；调用方传入数据库中的全部设备 ID，既能删除共享 L2， + * 又能通过 syncLocal 让其他实例立即移除对应 L1，且不会误删同 Redis 中的其他业务数据。

+ */ + public void invalidateAllRules(Collection deviceIds) { + clearSpring("normalizeRule"); + metrics.invalidation("local"); + if (deviceIds == null) return; + for (Long deviceId : deviceIds) { + if (deviceId == null) continue; + jetCacheInvalidator.rulesForDevice(deviceId); + metrics.invalidation("remote"); + } + } + private void evictSpring(String cacheName, Object key) { Cache cache = springCacheManager.getCache(cacheName); if (cache != null) cache.evict(key); } + + private void clearSpring(String cacheName) { + Cache cache = springCacheManager.getCache(cacheName); + if (cache != null) cache.clear(); + } } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCompiledCacheInvalidationBus.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCompiledCacheInvalidationBus.java new file mode 100644 index 0000000..44268a0 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCompiledCacheInvalidationBus.java @@ -0,0 +1,85 @@ +package com.Modules.NormalData.cache; + +import com.Modules.NormalData.compiled.CompiledRuleCache; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Component; + +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.LinkedHashSet; +import java.util.Set; +import java.util.StringJoiner; + +/** 通过 Redis Pub/Sub 让所有消费节点同步清理 JVM 本地编译规则缓存。 */ +@Component +public class NormalizationCompiledCacheInvalidationBus implements MessageListener { + public static final String TOPIC = "xdr:norm:v1:compiled-rule:invalidate"; + private static final String ALL = "ALL"; + private static final String RULES_PREFIX = "RULES:"; + private static final Logger log = LoggerFactory.getLogger( + NormalizationCompiledCacheInvalidationBus.class); + + private final CompiledRuleCache compiledRuleCache; + private final StringRedisTemplate redisTemplate; + + public NormalizationCompiledCacheInvalidationBus(CompiledRuleCache compiledRuleCache, + StringRedisTemplate redisTemplate) { + this.compiledRuleCache = compiledRuleCache; + this.redisTemplate = redisTemplate; + } + + public void invalidateRulesEverywhere(Collection ruleIds) { + Set distinct = positiveIds(ruleIds); + if (distinct.isEmpty()) return; + for (Long ruleId : distinct) compiledRuleCache.invalidateRule(ruleId); + redisTemplate.convertAndSend(TOPIC, ruleMessage(distinct)); + } + + public void invalidateAllEverywhere() { + compiledRuleCache.invalidateAll(); + redisTemplate.convertAndSend(TOPIC, ALL); + } + + @Override + public void onMessage(Message message, byte[] pattern) { + if (message == null || message.getBody() == null) return; + String command = new String(message.getBody(), StandardCharsets.UTF_8).trim(); + if (ALL.equals(command)) { + compiledRuleCache.invalidateAll(); + return; + } + if (!command.startsWith(RULES_PREFIX)) { + log.warn("忽略未知的规则缓存失效广播: {}", command); + return; + } + String ids = command.substring(RULES_PREFIX.length()); + if (ids.isEmpty()) return; + for (String value : ids.split(",")) { + try { + long ruleId = Long.parseLong(value); + if (ruleId > 0L) compiledRuleCache.invalidateRule(ruleId); + } catch (NumberFormatException invalidId) { + log.warn("忽略非法的规则缓存失效广播参数: {}", value); + } + } + } + + private static Set positiveIds(Collection ruleIds) { + Set distinct = new LinkedHashSet<>(); + if (ruleIds == null) return distinct; + for (Long ruleId : ruleIds) { + if (ruleId != null && ruleId > 0L) distinct.add(ruleId); + } + return distinct; + } + + private static String ruleMessage(Collection ruleIds) { + StringJoiner values = new StringJoiner(",", RULES_PREFIX, ""); + for (Long ruleId : ruleIds) values.add(ruleId.toString()); + return values.toString(); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationRuleCacheManagementService.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationRuleCacheManagementService.java new file mode 100644 index 0000000..b8756b7 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationRuleCacheManagementService.java @@ -0,0 +1,83 @@ +package com.Modules.NormalData.cache; + +import com.common.mapper.DeviceDeviceMapper; +import com.common.mapper.DmNormalizeRuleMapper; +import org.springframework.stereotype.Service; + +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +/** 管理接口使用的规则缓存清理编排；不会清除设备、任务或其他业务 Redis 数据。 */ +@Service +public class NormalizationRuleCacheManagementService { + private final NormalizationCacheInvalidationService invalidationService; + private final NormalizationCompiledCacheInvalidationBus compiledCacheInvalidationBus; + private final DeviceDeviceMapper deviceMapper; + private final DmNormalizeRuleMapper ruleMapper; + + public NormalizationRuleCacheManagementService( + NormalizationCacheInvalidationService invalidationService, + NormalizationCompiledCacheInvalidationBus compiledCacheInvalidationBus, + DeviceDeviceMapper deviceMapper, + DmNormalizeRuleMapper ruleMapper) { + this.invalidationService = invalidationService; + this.compiledCacheInvalidationBus = compiledCacheInvalidationBus; + this.deviceMapper = deviceMapper; + this.ruleMapper = ruleMapper; + } + + /** 清理一个设备的规则快照，并移除该设备当前关联规则的实体和编译缓存。 */ + public ClearResult clearDevice(long deviceId) { + if (deviceId <= 0L) throw new IllegalArgumentException("deviceId 必须大于 0"); + Set ruleIds = new LinkedHashSet<>(); + addAllNonNull(ruleIds, nullSafe(ruleMapper.selectRuleIdsByDeviceId(deviceId))); + invalidationService.invalidateRulesForDevice(deviceId); + for (Long ruleId : ruleIds) { + invalidationService.invalidateRuleEntity(ruleId); + } + compiledCacheInvalidationBus.invalidateRulesEverywhere(ruleIds); + return new ClearResult("DEVICE", deviceId, 1, ruleIds.size()); + } + + /** 清理所有设备规则快照、规则实体和当前 JVM 的全部编译缓存。 */ + public ClearResult clearAll() { + Set deviceIds = new LinkedHashSet<>(); + addAllNonNull(deviceIds, deviceMapper.selectAllIds()); + // 即使存在历史孤立关联，也应删除其规则快照键。 + addAllNonNull(deviceIds, ruleMapper.selectAllAssociatedDeviceIds()); + invalidationService.invalidateAllRules(deviceIds); + compiledCacheInvalidationBus.invalidateAllEverywhere(); + return new ClearResult("ALL", null, deviceIds.size(), 0); + } + + private static List nullSafe(List values) { + return values == null ? Collections.emptyList() : values; + } + + private static void addAllNonNull(Set target, Collection values) { + if (values == null) return; + for (Long value : values) if (value != null) target.add(value); + } + + public static final class ClearResult { + private final String scope; + private final Long deviceId; + private final int deviceCount; + private final int associatedRuleCount; + + public ClearResult(String scope, Long deviceId, int deviceCount, int associatedRuleCount) { + this.scope = scope; + this.deviceId = deviceId; + this.deviceCount = deviceCount; + this.associatedRuleCount = associatedRuleCount; + } + + public String getScope() { return scope; } + public Long getDeviceId() { return deviceId; } + public int getDeviceCount() { return deviceCount; } + public int getAssociatedRuleCount() { return associatedRuleCount; } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java index f6d93c8..eebe46e 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java @@ -1,6 +1,5 @@ package com.Modules.NormalData.cache; -import com.Modules.NormalData.compiled.CompiledRuleCache; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; @@ -14,21 +13,21 @@ public class NormalizeRuleCacheInvalidationListener { private static final Logger log = LoggerFactory.getLogger(NormalizeRuleCacheInvalidationListener.class); private final NormalizationCacheInvalidationService invalidationService; - private final CompiledRuleCache compiledRuleCache; + private final NormalizationCompiledCacheInvalidationBus compiledCacheInvalidationBus; private final NormalizationRuleMetrics metrics; public NormalizeRuleCacheInvalidationListener(NormalizationCacheInvalidationService invalidationService, - CompiledRuleCache compiledRuleCache, + NormalizationCompiledCacheInvalidationBus compiledCacheInvalidationBus, NormalizationRuleMetrics metrics) { this.invalidationService = invalidationService; - this.compiledRuleCache = compiledRuleCache; + this.compiledCacheInvalidationBus = compiledCacheInvalidationBus; this.metrics = metrics; } @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) public void afterCommit(NormalizeRuleChangedEvent event) { metrics.invalidation("after_commit"); - // 先移除跨 JVM 的规则事实，再清除本 JVM 所有历史 fingerprint 编译对象。 + // 先移除跨 JVM 的规则事实，再广播清除所有 JVM 的历史 fingerprint 对象。 for (Long deviceId : event.getDeviceIds()) { try { invalidationService.invalidateRulesForDevice(deviceId); @@ -39,7 +38,8 @@ public class NormalizeRuleCacheInvalidationListener { } } try { - compiledRuleCache.invalidateRule(event.getRuleId()); + compiledCacheInvalidationBus.invalidateRulesEverywhere( + java.util.Collections.singleton(event.getRuleId())); } catch (RuntimeException cacheFailure) { metrics.invalidation("failure"); // 数据库事实已提交；raw rules 的短 TTL 是最终一致性兜底。 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCache.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCache.java index 808c1f8..13ac8f6 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCache.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCache.java @@ -192,6 +192,18 @@ public class CompiledRuleCache { metrics.compiledCacheGauge(compiled.estimatedSize(), evictions.sum()); } + /** 清空当前 JVM 中所有规则编译结果、匹配画像及短失败缓存。 */ + public void invalidateAll() { + compiled.invalidateAll(); + matchProfiles.invalidateAll(); + failures.invalidateAll(); + matchFailures.invalidateAll(); + loggedFailures.invalidateAll(); + keysByRule.invalidateAll(); + matchKeysByRule.invalidateAll(); + metrics.compiledCacheGauge(0L, evictions.sum()); + } + public long estimatedSize() { return compiled.estimatedSize(); } public long compileSuccessCount() { return compileSuccess.sum(); } public long compileFailureCount() { return compileFailure.sum(); } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DeviceDeviceMapper.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DeviceDeviceMapper.java index dfe4373..dc699ea 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DeviceDeviceMapper.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DeviceDeviceMapper.java @@ -21,6 +21,9 @@ public interface DeviceDeviceMapper { */ List selectAll(); + /** 规则缓存全量失效只需要主键，避免加载整行设备信息。 */ + List selectAllIds(); + /** * 根据IP地址查询设备 */ @@ -90,4 +93,4 @@ public interface DeviceDeviceMapper { * 根据厂商查询设备 */ List selectByVendor(String vendor); -} \ No newline at end of file +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DmNormalizeRuleMapper.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DmNormalizeRuleMapper.java index 9e8c945..884d836 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DmNormalizeRuleMapper.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/DmNormalizeRuleMapper.java @@ -24,6 +24,10 @@ public interface DmNormalizeRuleMapper { List selectDeviceIdsByRuleId(@Param("ruleId") Long ruleId); + List selectRuleIdsByDeviceId(@Param("deviceId") Long deviceId); + + List selectAllAssociatedDeviceIds(); + /** * 根据ID更新数据标准化规则 * @param rule 数据标准化规则对象 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/NormalizationCacheInvalidationRedisConfig.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/NormalizationCacheInvalidationRedisConfig.java new file mode 100644 index 0000000..98980dc --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/NormalizationCacheInvalidationRedisConfig.java @@ -0,0 +1,24 @@ +package com.config; + +import com.Modules.NormalData.cache.NormalizationCompiledCacheInvalidationBus; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; + +/** 规则编译缓存的跨实例失效广播通道。 */ +@Configuration +public class NormalizationCacheInvalidationRedisConfig { + + @Bean + public RedisMessageListenerContainer normalizationCacheInvalidationListenerContainer( + RedisConnectionFactory connectionFactory, + NormalizationCompiledCacheInvalidationBus listener) { + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + container.addMessageListener(listener, + new ChannelTopic(NormalizationCompiledCacheInvalidationBus.TOPIC)); + return container; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java index 19d4065..24e39f5 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java @@ -12,7 +12,7 @@ import org.springframework.stereotype.Component; public class SyslogBuildStartupLogger { // 确认包含此功能修订，不代表唯一构建；后续功能修订时同步更新此标识。 - static final String CODE_REVISION = "ASYNC_STATS_ADAPTIVE_RAW_V20260913_57"; + static final String CODE_REVISION = "ASYNC_STATS_ADAPTIVE_RAW_V20260917_63"; private final Environment environment; public SyslogBuildStartupLogger(Environment environment) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/NormalizationRuleCacheController.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/NormalizationRuleCacheController.java new file mode 100644 index 0000000..c79c9ad --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/NormalizationRuleCacheController.java @@ -0,0 +1,52 @@ +package com.controllers; + +import com.Modules.NormalData.cache.NormalizationRuleCacheManagementService; +import com.Modules.NormalData.cache.NormalizationRuleCacheManagementService.ClearResult; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** 只清理标准化规则相关缓存，不执行现有 /api/cache/all 的全 Redis 删除。 */ +@RestController +@RequestMapping("/api/syslog/normalization/cache") +public class NormalizationRuleCacheController { + private final NormalizationRuleCacheManagementService service; + + public NormalizationRuleCacheController(NormalizationRuleCacheManagementService service) { + this.service = service; + } + + @DeleteMapping("/devices/{deviceId}") + public ResponseEntity> clearDevice(@PathVariable Long deviceId) { + if (deviceId == null || deviceId <= 0L) { + Map error = new LinkedHashMap<>(); + error.put("cleared", false); + error.put("errorCode", "INVALID_DEVICE_ID"); + error.put("message", "deviceId 必须大于 0"); + return ResponseEntity.badRequest().body(error); + } + return ResponseEntity.ok(response(service.clearDevice(deviceId))); + } + + @DeleteMapping + public Map clearAll() { + return response(service.clearAll()); + } + + private Map response(ClearResult result) { + Map body = new LinkedHashMap<>(); + body.put("cleared", true); + body.put("scope", result.getScope()); + if (result.getDeviceId() != null) body.put("deviceId", result.getDeviceId()); + body.put("deviceCount", result.getDeviceCount()); + if (result.getDeviceId() != null) { + body.put("associatedRuleCount", result.getAssociatedRuleCount()); + } + return body; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/DeviceDeviceMapper.xml b/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/DeviceDeviceMapper.xml index af4e296..cb8c7aa 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/DeviceDeviceMapper.xml +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/DeviceDeviceMapper.xml @@ -70,6 +70,10 @@ FROM device_device + + + + + + - \ No newline at end of file + + UPDATE device_device + SET collection_mode = #{collectionMode,javaType=com.common.entity.DeviceCollectionMode,typeHandler=org.apache.ibatis.type.EnumTypeHandler}, + updated_at = SYSDATE, + update_time = SYSDATE + WHERE id = #{id} + AND deleted_at IS NULL + AND (del_flag IS NULL OR del_flag = '0') + + + diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/common/controller/DeviceDeviceControllerTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/common/controller/DeviceDeviceControllerTest.java new file mode 100644 index 0000000..ffa3e89 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/common/controller/DeviceDeviceControllerTest.java @@ -0,0 +1,131 @@ +package com.common.controller; + +import com.common.entity.DeviceCollectionMode; +import com.common.entity.DeviceDevice; +import com.common.service.DeviceDeviceService; +import com.haobang.dispatch.DeviceContextResolver; +import org.junit.jupiter.api.Test; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.put; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +class DeviceDeviceControllerTest { + @Test + void rejectsInvalidCollectionMode() throws Exception { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + MockMvc mvc = MockMvcBuilders.standaloneSetup( + new DeviceDeviceController(service, resolver)).build(); + + mvc.perform(put("/api/devices/40/collection-mode") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"collectionMode\":\"INVALID\"}")) + .andExpect(status().isBadRequest()); + + verify(service, never()).updateCollectionMode( + org.mockito.ArgumentMatchers.anyInt(), org.mockito.ArgumentMatchers.any()); + } + + @Test + void rejectsAttemptsToUpdateOtherDeviceFields() throws Exception { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + MockMvc mvc = MockMvcBuilders.standaloneSetup( + new DeviceDeviceController(service, resolver)).build(); + + mvc.perform(put("/api/devices/40/collection-mode") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"collectionMode\":\"API_ONLY\",\"name\":\"tampered\"}")) + .andExpect(status().isBadRequest()); + + verify(service, never()).updateCollectionMode( + org.mockito.ArgumentMatchers.anyInt(), org.mockito.ArgumentMatchers.any()); + } + + @Test + void updateInvalidatesLocalCachesAndRefreshesIngressSnapshot() throws Exception { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + DeviceDevice updated = new DeviceDevice(); + updated.setId(40); + updated.setIp("10.56.171.164"); + updated.setCollectionMode(DeviceCollectionMode.API_ONLY); + when(service.updateCollectionMode(40, DeviceCollectionMode.API_ONLY)).thenReturn(updated); + MockMvc mvc = MockMvcBuilders.standaloneSetup( + new DeviceDeviceController(service, resolver)).build(); + + mvc.perform(put("/api/devices/40/collection-mode") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"collectionMode\":\"API_ONLY\"}")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.id").value(40)) + .andExpect(jsonPath("$.collectionMode").value("API_ONLY")); + + verify(resolver).invalidateAll(); + verify(resolver).refreshIngressSnapshot(); + } + + @Test + void returnsNotFoundForMissingOrDeletedDevice() throws Exception { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + MockMvc mvc = MockMvcBuilders.standaloneSetup( + new DeviceDeviceController(service, resolver)).build(); + + mvc.perform(put("/api/devices/404/collection-mode") + .contentType(MediaType.APPLICATION_JSON) + .content("{\"collectionMode\":\"API_ONLY\"}")) + .andExpect(status().isNotFound()); + + verify(resolver, never()).refreshIngressSnapshot(); + } + + @Test + void clearsAllDeviceCacheLayersAndRefreshesSnapshot() throws Exception { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + MockMvc mvc = MockMvcBuilders.standaloneSetup( + new DeviceDeviceController(service, resolver)).build(); + + mvc.perform(delete("/api/devices/cache")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.cleared").value(true)) + .andExpect(jsonPath("$.scope").value("ALL")); + + verify(service).evictAllDeviceCaches(); + verify(resolver).invalidateAll(); + verify(resolver).refreshIngressSnapshot(); + } + + @Test + void clearsSpecifiedDeviceCacheLayersAndRefreshesSnapshot() throws Exception { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + DeviceDevice device = new DeviceDevice(); + device.setId(40); + device.setIp("10.56.171.164"); + when(service.evictDeviceCaches(40)).thenReturn(device); + MockMvc mvc = MockMvcBuilders.standaloneSetup( + new DeviceDeviceController(service, resolver)).build(); + + mvc.perform(delete("/api/devices/40/cache")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.cleared").value(true)) + .andExpect(jsonPath("$.scope").value("DEVICE")) + .andExpect(jsonPath("$.deviceId").value(40)) + .andExpect(jsonPath("$.ip").value("10.56.171.164")); + + verify(service).evictDeviceCaches(40); + verify(resolver).invalidateDevice(40, "10.56.171.164"); + verify(resolver).refreshIngressSnapshot(); + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/common/mapper/DeviceDeviceMapperXmlTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/common/mapper/DeviceDeviceMapperXmlTest.java new file mode 100644 index 0000000..8576287 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/common/mapper/DeviceDeviceMapperXmlTest.java @@ -0,0 +1,40 @@ +package com.common.mapper; + +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import org.apache.ibatis.builder.xml.XMLMapperBuilder; +import org.apache.ibatis.io.Resources; +import org.apache.ibatis.mapping.ResultMapping; +import org.apache.ibatis.session.Configuration; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +class DeviceDeviceMapperXmlTest { + @Test + void parsesCollectionModeMappingAndDedicatedUpdateStatement() throws Exception { + Configuration configuration = new Configuration(); + String resource = "mapper/DeviceDeviceMapper.xml"; + try (InputStream input = Resources.getResourceAsStream(resource)) { + new XMLMapperBuilder(input, configuration, resource, configuration.getSqlFragments()).parse(); + } + + ResultMapping collectionMode = configuration + .getResultMap("com.common.mapper.DeviceDeviceMapper.BaseResultMap") + .getResultMappings().stream() + .filter(mapping -> "collectionMode".equals(mapping.getProperty())) + .findFirst().orElseThrow(() -> new AssertionError("collectionMode mapping missing")); + assertTrue("collection_mode".equalsIgnoreCase(collectionMode.getColumn())); + + Map parameters = new HashMap<>(); + parameters.put("id", 40); + parameters.put("collectionMode", "API_ONLY"); + String sql = configuration + .getMappedStatement("com.common.mapper.DeviceDeviceMapper.updateCollectionMode") + .getBoundSql(parameters).getSql().replaceAll("\\s+", " ").toLowerCase(); + assertTrue(sql.contains("set collection_mode = ?")); + assertTrue(sql.contains("where id = ?")); + assertTrue(sql.contains("deleted_at is null")); + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/common/service/DeviceDeviceServiceImplTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/common/service/DeviceDeviceServiceImplTest.java new file mode 100644 index 0000000..7e4e11f --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/common/service/DeviceDeviceServiceImplTest.java @@ -0,0 +1,79 @@ +package com.common.service; + +import com.common.entity.DeviceCollectionMode; +import com.common.entity.DeviceDevice; +import com.common.mapper.DeviceDeviceMapper; +import com.common.service.impl.DeviceDeviceServiceImpl; +import com.haobang.util.SafeCacheUtil; +import org.junit.jupiter.api.Test; +import org.springframework.cache.Cache; +import org.springframework.cache.CacheManager; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class DeviceDeviceServiceImplTest { + @Test + void updateEvictsRawRedisKeysAndEntireSpringDeviceCache() { + DeviceDeviceMapper mapper = mock(DeviceDeviceMapper.class); + SafeCacheUtil safeCache = mock(SafeCacheUtil.class); + CacheManager cacheManager = mock(CacheManager.class); + Cache springCache = mock(Cache.class); + DeviceDevice current = new DeviceDevice(); + current.setId(40); + current.setIp("10.56.171.164"); + current.setDelFlag("0"); + DeviceDevice updated = new DeviceDevice(); + updated.setId(40); + updated.setIp(current.getIp()); + updated.setCollectionMode(DeviceCollectionMode.API_ONLY); + when(mapper.selectById(40)).thenReturn(current, updated); + when(mapper.updateCollectionMode(40, DeviceCollectionMode.API_ONLY)).thenReturn(1); + when(cacheManager.getCache("device")).thenReturn(springCache); + DeviceDeviceServiceImpl service = new DeviceDeviceServiceImpl(mapper, safeCache, cacheManager); + + DeviceDevice result = service.updateCollectionMode(40, DeviceCollectionMode.API_ONLY); + + assertEquals(DeviceCollectionMode.API_ONLY, result.getCollectionMode()); + verify(safeCache).evict("device:id:40", "device:ip:10.56.171.164"); + verify(springCache).clear(); + } + + @Test + void evictsOnlySpecifiedDeviceIdAndIpEntries() { + DeviceDeviceMapper mapper = mock(DeviceDeviceMapper.class); + SafeCacheUtil safeCache = mock(SafeCacheUtil.class); + CacheManager cacheManager = mock(CacheManager.class); + Cache springCache = mock(Cache.class); + DeviceDevice device = new DeviceDevice(); + device.setId(40); + device.setIp("10.56.171.164"); + when(mapper.selectById(40)).thenReturn(device); + when(cacheManager.getCache("device")).thenReturn(springCache); + DeviceDeviceServiceImpl service = new DeviceDeviceServiceImpl(mapper, safeCache, cacheManager); + + DeviceDevice result = service.evictDeviceCaches(40); + + assertEquals(40, result.getId()); + verify(safeCache).evict("device:id:40", "device:ip:10.56.171.164"); + verify(springCache).evict("device:id:40"); + verify(springCache).evict("device:ip:10.56.171.164"); + } + + @Test + void evictsAllRawRedisAndSpringDeviceCaches() { + DeviceDeviceMapper mapper = mock(DeviceDeviceMapper.class); + SafeCacheUtil safeCache = mock(SafeCacheUtil.class); + CacheManager cacheManager = mock(CacheManager.class); + Cache springCache = mock(Cache.class); + when(cacheManager.getCache("device")).thenReturn(springCache); + DeviceDeviceServiceImpl service = new DeviceDeviceServiceImpl(mapper, safeCache, cacheManager); + + service.evictAllDeviceCaches(); + + verify(safeCache).evictMatching("device:id:*", "device:ip:*"); + verify(springCache).clear(); + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/config/SyslogBuildStartupLoggerTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/config/SyslogBuildStartupLoggerTest.java new file mode 100644 index 0000000..33d8aff --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/config/SyslogBuildStartupLoggerTest.java @@ -0,0 +1,57 @@ +package com.config; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.springframework.mock.env.MockEnvironment; + +class SyslogBuildStartupLoggerTest { + @Test + void logsCodeRevisionAndEffectiveOverrides() { + MockEnvironment environment = new MockEnvironment() + .withProperty("device.api.enabled", "false") + .withProperty("device.api.tls-verify-enabled", "true") + .withProperty("app.service.device_collect_id", "40") + .withProperty("spring.kafka.producer.topic", "prod-syslog"); + environment.setActiveProfiles("prod"); + SyslogBuildStartupLogger reporter = new SyslogBuildStartupLogger(environment); + Logger logger = (Logger) LoggerFactory.getLogger(SyslogBuildStartupLogger.class); + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + try { + reporter.onApplicationReady(); + assertEquals(1, appender.list.size()); + String message = appender.list.get(0).getFormattedMessage(); + assertTrue(message.contains("[SYSLOG_BUILD_READY] codeRevision=" + + SyslogBuildStartupLogger.CODE_REVISION)); + assertTrue(message.contains("application=syslog-serve")); + assertTrue(message.contains("profiles=prod")); + assertTrue(message.contains("apiPullEnabled=false")); + assertTrue(message.contains("apiTlsVerifyEnabled=true")); + assertTrue(message.contains("deviceCollectId=40")); + assertTrue(message.contains("kafkaTopic=prod-syslog")); + } finally { + logger.detachAppender(appender); + appender.stop(); + } + } + + @Test + void usesSafeDefaultsAndIgnoresConfiguredRevision() { + String summary = new SyslogBuildStartupLogger(new MockEnvironment() + .withProperty("codeRevision", "fake")).startupSummary(); + assertTrue(summary.contains("codeRevision=" + SyslogBuildStartupLogger.CODE_REVISION)); + assertFalse(summary.contains("codeRevision=fake")); + assertTrue(summary.contains("syslogEnabled=true")); + assertTrue(summary.contains("apiPullEnabled=true")); + assertTrue(summary.contains("apiTlsVerifyEnabled=false")); + assertTrue(summary.contains("walEnabled=true")); + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullPropertiesTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullPropertiesTest.java new file mode 100644 index 0000000..7bedfd5 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullPropertiesTest.java @@ -0,0 +1,15 @@ +package com.haobang.apipull; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class ApiPullPropertiesTest { + @Test + void apiPullIsEnabledByDefault() { + ApiPullProperties properties = new ApiPullProperties(); + assertTrue(properties.isEnabled()); + assertFalse(properties.isTlsVerifyEnabled()); + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullRunnerTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullRunnerTest.java index c029a6a..3a80937 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullRunnerTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullRunnerTest.java @@ -154,6 +154,44 @@ class ApiPullRunnerTest { verify(fixture.ingress, times(3)).tryAppend(any(IngressEvent.class)); } + @Test + void unpagedAdapterSplitsUntilSingleResponseContainsWholeWindow() { + RunnerFixture fixture = new RunnerFixture(); + fixture.config.setPageSize(2); + fixture.config.setMaxPagesPerRun(1); + + when(fixture.adapter.pull(any())) + .thenAnswer( + invocation -> { + PullRequest request = invocation.getArgument(0); + long windowSeconds = + Duration.between(WINDOW_START, request.getQueryEnd()).getSeconds(); + return windowSeconds == 60L ? page(3L, 2, 0L) : page(2L, 2, 0L); + }); + + AtomicReference committedEnd = new AtomicReference<>(); + when(fixture.mapper.commitCheckpoint( + anyInt(), anyLong(), anyString(), anyLong(), anyLong(), anyString(), any(), any(), any())) + .thenAnswer( + invocation -> { + committedEnd.set(invocation.getArgument(6)); + return 1; + }); + + fixture.runner.execute(fixture.config, fixture.checkpoint, fixture.run); + + assertEquals("SUCCESS", fixture.run.getStatus()); + assertEquals(Integer.valueOf(1), fixture.run.getPagesFetched()); + assertEquals(Long.valueOf(2L), fixture.run.getRecordsDurable()); + assertEquals(ApiPullTime.toDatabase(WINDOW_START.plusSeconds(30L)), committedEnd.get()); + org.mockito.ArgumentCaptor requests = + org.mockito.ArgumentCaptor.forClass(PullRequest.class); + verify(fixture.adapter, times(2)).pull(requests.capture()); + assertEquals(0L, requests.getAllValues().get(0).getOffset()); + assertEquals(0L, requests.getAllValues().get(1).getOffset()); + verify(fixture.ingress, times(2)).tryAppend(any(IngressEvent.class)); + } + @Test void schedulingSliceCanSplitBelowTheHardRunCapacity() { RunnerFixture fixture = new RunnerFixture(); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullSchedulerTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullSchedulerTest.java index 4165567..2d672c4 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullSchedulerTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullSchedulerTest.java @@ -1,6 +1,12 @@ package com.haobang.apipull; +import com.common.entity.DeviceCollectionMode; +import com.common.entity.DeviceDevice; +import com.common.service.DeviceDeviceService; +import com.common.service.DeviceUnknownService; import com.common.mapper.DeviceCollectApiMapper; +import com.haobang.config.AppProperties; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.apipull.persistence.DeviceCollectApiCheckpoint; import com.haobang.apipull.persistence.DeviceCollectApiConfig; import com.haobang.wal.SyslogWalService; @@ -19,12 +25,33 @@ class ApiPullSchedulerTest { DeviceCollectApiCheckpoint cp=new DeviceCollectApiCheckpoint();cp.setFencingToken(1L);when(mapper.selectCheckpoint(1)).thenReturn(cp); CountDownLatch entered=new CountDownLatch(1);CountDownLatch release=new CountDownLatch(1); doAnswer(i->{entered.countDown();release.await(2,TimeUnit.SECONDS);return null;}).when(runner).execute(eq(first),eq(cp),any()); - ApiPullScheduler scheduler=new ApiPullScheduler(p,mapper,runner,wal);scheduler.init(); + ApiPullScheduler scheduler=new ApiPullScheduler(p,mapper,runner,wal,resolver(DeviceCollectionMode.BOTH));scheduler.init(); try{assertNotNull(scheduler.trigger(1));assertTrue(entered.await(1,TimeUnit.SECONDS)); ApiPullException conflict=assertThrows(ApiPullException.class,()->scheduler.trigger(2));assertEquals("LEASE_CONFLICT",conflict.getCode()); verify(mapper,never()).acquireLease(eq(2),anyLong(),anyString(),anyInt());} finally{release.countDown();scheduler.shutdown();} } + + @Test void syslogOnlyScheduledTaskAdvancesNextRunWithoutLeaseOrHttp(){ + ApiPullProperties p=new ApiPullProperties();p.setEnabled(true);p.setInstanceId("i1");p.setScanIntervalMs(5000);p.setLeaseSeconds(120); + p.setGlobalConcurrency(1);p.setPerHostConcurrency(1);p.setSlowPageThresholdMs(1000);p.setBackpressureTimeoutMs(1000); + DeviceCollectApiMapper mapper=mock(DeviceCollectApiMapper.class);ApiPullRunner runner=mock(ApiPullRunner.class);SyslogWalService wal=mock(SyslogWalService.class); + DeviceCollectApiConfig config=config(1,"https://localhost/ndr/api"); + when(mapper.selectDueConfigs()).thenReturn(java.util.Collections.singletonList(config)); + ApiPullScheduler scheduler=new ApiPullScheduler(p,mapper,runner,wal,resolver(DeviceCollectionMode.SYSLOG_ONLY));scheduler.init(); + try{scheduler.scan(); + verify(mapper).scheduleNext(1,1L,10); + verify(mapper,never()).acquireLease(anyInt(),anyLong(),anyString(),anyInt()); + verifyNoInteractions(runner); + }finally{scheduler.shutdown();} + } + + private DeviceContextResolver resolver(DeviceCollectionMode mode){ + DeviceDeviceService devices=mock(DeviceDeviceService.class); + when(devices.getByIdSafely(anyInt())).thenAnswer(i->{DeviceDevice d=new DeviceDevice();d.setId(i.getArgument(0));d.setDeviceCollectId(1);d.setCollectionMode(mode);return d;}); + AppProperties app=new AppProperties();app.setAppServiceDeviceCollectId(1);app.setSyslogDeviceContextCacheTtlMs(60000L); + return new DeviceContextResolver(devices,mock(DeviceUnknownService.class),app); + } private DeviceCollectApiConfig config(int id,String base){DeviceCollectApiConfig c=new DeviceCollectApiConfig();c.setCollectTaskId(id);c.setEnabled(1);c.setConfigVersion(1L); c.setBaseUrl(base);c.setPollIntervalSeconds(10);c.setDeviceId(id);c.setAdapterType("NDRSERVER_V1");c.setAdapterVersion(1);c.setStreamKey("NDR_ALERT");return c;} } diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullTaskServiceTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullTaskServiceTest.java index 74e4897..a1632d6 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullTaskServiceTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullTaskServiceTest.java @@ -2,13 +2,16 @@ package com.haobang.apipull; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import com.common.entity.DeviceCollectTask; +import com.common.entity.DeviceCollectionMode; import com.common.entity.DeviceDevice; import com.common.mapper.DeviceCollectApiMapper; import com.common.mapper.DeviceCollectTaskMapper; @@ -157,6 +160,11 @@ class ApiPullTaskServiceTest { (credentialRef, adapterType, deviceId) -> new CredentialSnapshot(1, "v1", adapterType, Collections.emptyMap()); AppProperties app = new AppProperties(); + app.setAppServiceDeviceCollectId(7); + DeviceDevice allowed = new DeviceDevice(); + allowed.setId(126); + allowed.setDeviceCollectId(7); + when(devices.getByIdSafely(126)).thenReturn(allowed); ApiPullProperties apiProperties = new ApiPullProperties(); apiProperties.setAllowedHosts("localhost"); ApiPullTaskService service = @@ -178,12 +186,75 @@ class ApiPullTaskServiceTest { .collect(Collectors.toMap(ValidationCheckResponse::getName, value -> value)); assertEquals("PASS", checks.get("TLS_HANDSHAKE").getStatus()); + assertEquals( + "TLS 连接通过（全局已关闭证书链和主机名校验）", + checks.get("TLS_HANDSHAKE").getSummary()); assertEquals("PASS", checks.get("AUTHENTICATION").getStatus()); assertEquals("PASS", checks.get("API_PERMISSION").getStatus()); assertEquals("FAIL", checks.get("MINIMAL_ALERT_QUERY").getStatus()); assertEquals("QUERY_RESPONSE_SCHEMA_ERROR", checks.get("MINIMAL_ALERT_QUERY").getErrorCode()); } + @Test + void syslogOnlyDeviceCannotCreateApiPullTask() { + DeviceCollectTaskMapper tasks = mock(DeviceCollectTaskMapper.class); + DeviceCollectApiMapper apiMapper = mock(DeviceCollectApiMapper.class); + DeviceDeviceService devices = mock(DeviceDeviceService.class); + DeviceDevice device = new DeviceDevice(); + device.setId(126); + device.setDeviceCollectId(7); + device.setCollectionMode(DeviceCollectionMode.SYSLOG_ONLY); + when(devices.getByIdSafely(126)).thenReturn(device); + AppProperties app = new AppProperties(); + app.setAppServiceDeviceCollectId(7); + ApiPullProperties properties = new ApiPullProperties(); + properties.setAllowedHosts("localhost"); + ApiPullTaskService service = new ApiPullTaskService( + tasks, apiMapper, devices, mock(AdapterRegistry.class), + mock(CredentialProvider.class), new SafeBaseUrlValidator(properties), + mock(ApiPullScheduler.class), app, properties, new ObjectMapper()); + + ApiPullException failure = assertThrows(ApiPullException.class, () -> service.create(request())); + + assertEquals("COLLECTION_MODE_DENIED", failure.getCode()); + verify(tasks, never()).insertApiPullTask(any()); + verify(apiMapper, never()).insertConfig(any()); + } + + @Test + void syslogOnlyDeviceCannotValidateResumeOrManuallyTriggerApiTask() { + DeviceCollectApiMapper apiMapper = mock(DeviceCollectApiMapper.class); + DeviceCollectApiConfig config = new DeviceCollectApiConfig(); + config.setCollectTaskId(12); + config.setDeviceId(126); + config.setConfigVersion(3L); + when(apiMapper.selectConfig(12)).thenReturn(config); + DeviceDeviceService devices = mock(DeviceDeviceService.class); + DeviceDevice device = new DeviceDevice(); + device.setId(126); + device.setDeviceCollectId(7); + device.setCollectionMode(DeviceCollectionMode.SYSLOG_ONLY); + when(devices.getByIdSafely(126)).thenReturn(device); + AppProperties app = new AppProperties(); + app.setAppServiceDeviceCollectId(7); + ApiPullProperties properties = new ApiPullProperties(); + AdapterRegistry adapters = mock(AdapterRegistry.class); + ApiPullScheduler scheduler = mock(ApiPullScheduler.class); + ApiPullTaskService service = new ApiPullTaskService( + mock(DeviceCollectTaskMapper.class), apiMapper, devices, adapters, + mock(CredentialProvider.class), new SafeBaseUrlValidator(properties), scheduler, + app, properties, new ObjectMapper()); + + assertEquals("COLLECTION_MODE_DENIED", + assertThrows(ApiPullException.class, () -> service.validate(12, 3L)).getCode()); + assertEquals("COLLECTION_MODE_DENIED", + assertThrows(ApiPullException.class, () -> service.resume(12, 3L)).getCode()); + assertEquals("COLLECTION_MODE_DENIED", + assertThrows(ApiPullException.class, () -> service.trigger(12, 3L)).getCode()); + verifyNoInteractions(adapters, scheduler); + verify(apiMapper, never()).insertRun(any()); + } + @Test void updateFromOldClientKeepsExistingSchedulerPriority() { DeviceCollectTaskMapper tasks = mock(DeviceCollectTaskMapper.class); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapterTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapterTest.java index 9021593..01a01af 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapterTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapterTest.java @@ -29,6 +29,15 @@ import org.apache.http.util.EntityUtils; import org.junit.jupiter.api.Test; class NdrServerApiAdapterTest { + @Test + void httpClientCanExplicitlyDisableOrKeepTlsVerification() throws Exception { + try (CloseableHttpClient verified = NdrServerApiAdapter.buildHttpClient(true); + CloseableHttpClient internalInsecure = NdrServerApiAdapter.buildHttpClient(false)) { + assertNotNull(verified); + assertNotNull(internalInsecure); + } + } + @Test void challengeAnswerUsesExactUtf8AndLowercaseMd5() throws Exception { assertEquals( diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerSpringWiringTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerSpringWiringTest.java index 0225125..66ff85a 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerSpringWiringTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/ndr/NdrServerSpringWiringTest.java @@ -31,6 +31,7 @@ class NdrServerSpringWiringTest { }; context.registerBean(CredentialProvider.class, () -> credentialProvider); + context.registerBean(ApiPullProperties.class, () -> properties); context.registerBean(SafeBaseUrlValidator.class, () -> new SafeBaseUrlValidator(properties)); context.registerBean(ObjectMapper.class, () -> new ObjectMapper()); context.register(NdrServerApiAdapter.class); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapterTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapterTest.java index ed1f105..83f34a3 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapterTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapterTest.java @@ -30,6 +30,14 @@ class QuanxiApiAdapterTest { static final Instant END = START.plusSeconds(86400); static final String TOKEN = "synthetic-unit-test-token"; + @Test void supportsVerifiedAndGlobalInsecureTlsClients() throws Exception { + try (CloseableHttpClient verified = QuanxiApiAdapter.buildHttpClient(true); + CloseableHttpClient internalInsecure = QuanxiApiAdapter.buildHttpClient(false)) { + assertNotNull(verified); + assertNotNull(internalInsecure); + } + } + @Test void realFixtureBecomesTwentyIndependentOriginalDocumentsAndGoldenIds() throws Exception { Fixture f = new Fixture(); PullPage page = f.pull(); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiManagementTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiManagementTest.java index 0d4d2e4..2c49499 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiManagementTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/quanxi/QuanxiManagementTest.java @@ -7,6 +7,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.haobang.apipull.*; import com.haobang.apipull.adapter.*; import com.haobang.apipull.adapter.ndr.NdrServerApiAdapter; +import com.haobang.apipull.adapter.weibu.WeibuApiAdapter; import com.haobang.apipull.api.ApiPullDtos.*; import com.haobang.apipull.credential.*; import com.haobang.apipull.persistence.*; @@ -22,17 +23,25 @@ import static org.mockito.Mockito.*; import static org.mockito.ArgumentMatchers.*; class QuanxiManagementTest { - @Test void productionConstructorsRegisterBothVendorsWithoutExternalSystems() { + @Test void productionConstructorsRegisterAllVendorsWithoutExternalSystems() { try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) { ApiPullProperties props = new ApiPullProperties(); props.setAllowedHosts("192.0.2.10"); + context.registerBean(ApiPullProperties.class, () -> props); context.registerBean(CredentialProvider.class, () -> mock(CredentialProvider.class)); context.registerBean(ObjectMapper.class, () -> new ObjectMapper()); context.registerBean(SafeBaseUrlValidator.class, () -> new SafeBaseUrlValidator(props)); - context.register(QuanxiApiAdapter.class, NdrServerApiAdapter.class, AdapterRegistry.class); + context.register( + QuanxiApiAdapter.class, + NdrServerApiAdapter.class, + WeibuApiAdapter.class, + AdapterRegistry.class); context.refresh(); AdapterRegistry registry = context.getBean(AdapterRegistry.class); assertEquals(Collections.singleton(QuanxiApiAdapter.STREAM), registry.require("QUANXI_V1").streams()); assertTrue(registry.require("NDRSERVER_V1").streams().contains("NDR_ALERT")); + assertEquals( + Collections.singleton(WeibuApiAdapter.LOG), + registry.require(WeibuApiAdapter.TYPE).streams()); } } @@ -75,11 +84,15 @@ class QuanxiManagementTest { f.config.setConfigVersion(1L); f.config.setAdapterVersion(1); DeviceCollectApiMapper mapper = mock(DeviceCollectApiMapper.class); when(mapper.selectConfig(11)).thenReturn(f.config); + DeviceDeviceService devices = mock(DeviceDeviceService.class); + DeviceDevice device = new DeviceDevice(); device.setId(f.config.getDeviceId()); device.setDeviceCollectId(3); + when(devices.getByIdSafely(f.config.getDeviceId())).thenReturn(device); + AppProperties app = new AppProperties(); app.setAppServiceDeviceCollectId(3); ApiPullProperties props = new ApiPullProperties(); props.setAllowedHosts("192.0.2.10"); CredentialProvider credentials = (ref, type, id) -> new CredentialSnapshot(1, "fixture", type, Collections.emptyMap()); ApiPullTaskService service = new ApiPullTaskService(mock(DeviceCollectTaskMapper.class), mapper, - mock(DeviceDeviceService.class), new AdapterRegistry(Collections.singletonList(f.adapter)), - credentials, new SafeBaseUrlValidator(props), mock(ApiPullScheduler.class), new AppProperties(), props, new ObjectMapper()); + devices, new AdapterRegistry(Collections.singletonList(f.adapter)), + credentials, new SafeBaseUrlValidator(props), mock(ApiPullScheduler.class), app, props, new ObjectMapper()); ValidateApiPullTaskResponse response = service.validate(11, 1L); assertFalse(response.isValid()); Map checks = new HashMap<>(); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapterTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapterTest.java new file mode 100644 index 0000000..d31c57f --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapterTest.java @@ -0,0 +1,358 @@ +package com.haobang.apipull.adapter.weibu; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.haobang.apipull.ApiPullProperties; +import com.haobang.apipull.adapter.AdapterException; +import com.haobang.apipull.adapter.PullPage; +import com.haobang.apipull.adapter.PullRequest; +import com.haobang.apipull.credential.CredentialException; +import com.haobang.apipull.credential.CredentialProvider; +import com.haobang.apipull.credential.CredentialSchemas; +import com.haobang.apipull.credential.CredentialSnapshot; +import com.haobang.apipull.credential.SecretValue; +import com.haobang.apipull.persistence.DeviceCollectApiConfig; +import com.haobang.apipull.security.SafeBaseUrlValidator; +import java.io.ByteArrayInputStream; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.http.HttpEntity; +import org.apache.http.NameValuePair; +import org.apache.http.StatusLine; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.client.utils.URLEncodedUtils; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.util.EntityUtils; +import org.junit.jupiter.api.Test; + +/** 离线协议测试：不访问真实设备、数据库、Kafka 或真实凭据。 */ +class WeibuApiAdapterTest { + private static final ObjectMapper JSON = new ObjectMapper(); + private static final String API_KEY = "synthetic-api-key"; + private static final String SECRET = "synthetic-secret"; + private static final Instant START = Instant.parse("2026-09-16T00:00:00Z"); + private static final Instant END = START.plusSeconds(3600); + + @Test + void documentedHmacSha256UrlSafeSignatureIsExact() throws Exception { + assertEquals( + "giV2UIgRTNvcaVczQ2ifRtqZrLKZoFcoinMbWlTsCzQ", + WeibuApiAdapter.sign("api-key", "secret".toCharArray(), 1700000000L)); + } + + @Test + void httpClientCanExplicitlyDisableOrKeepTlsVerification() throws Exception { + try (CloseableHttpClient verified = WeibuApiAdapter.buildHttpClient(true); + CloseableHttpClient internalInsecure = WeibuApiAdapter.buildHttpClient(false)) { + assertNotNull(verified); + assertNotNull(internalInsecure); + } + } + + @Test + void requestUsesSignedQueryAndDocumentedLogSearchShape() throws Exception { + Fixture f = new Fixture(); + PullPage page = f.pull(0L); + + HttpPost request = f.request.get(); + assertEquals(WeibuApiAdapter.PATH, request.getURI().getPath()); + Map query = query(request); + assertEquals(API_KEY, query.get("api_key")); + long timestamp = Long.parseLong(query.get("auth_timestamp")); + assertEquals( + WeibuApiAdapter.sign(API_KEY, SECRET.toCharArray(), timestamp), query.get("sign")); + assertEquals(3, query.size()); + assertFalse(request.getConfig().isRedirectsEnabled()); + + JsonNode body = f.body.get(); + assertEquals(START.getEpochSecond(), body.path("time_from").asLong()); + assertEquals(END.getEpochSecond() - 1L, body.path("time_to").asLong()); + assertEquals(2, body.path("size").asInt()); + assertTrue(body.path("assets_group").isArray()); + assertFalse(body.path("search_connection").asBoolean()); + assertEquals("attack", body.path("net_data_type").get(0).asText()); + assertEquals("risk", body.path("net_data_type").get(1).asText()); + assertEquals("", body.path("sql").asText()); + assertTrue(body.path("columns").isArray()); + assertFalse(body.path("asc").asBoolean()); + assertFalse(body.has("condition")); + assertFalse(body.has("page")); + + assertEquals(2, page.getEvents().size()); + assertEquals(2L, page.getTotal()); + assertEquals("2", page.getDatasetFingerprint()); + assertEquals("log-1", page.getEvents().get(0).getExternalEventId()); + assertEquals(START.plusSeconds(100), page.getEvents().get(0).getEventTime()); + assertEquals(f.items().get(0).toString(), page.getEvents().get(0).getRawMessage()); + } + + @Test + void stableLogIdIsIdempotentAndOffsetPaginationIsRejected() throws Exception { + Fixture f = new Fixture(); + String stable = f.pull(0L).getEvents().get(0).getEventId(); + assertEquals(stable, f.pull(0L).getEvents().get(0).getEventId()); + ((ObjectNode) f.items().get(0)).put("machine", "192.0.2.99"); + assertEquals(stable, f.pull(0L).getEvents().get(0).getEventId()); + + assertEquals( + "REQUEST_RANGE_INVALID", + assertThrows(AdapterException.class, () -> f.pull(2L)).getCode()); + } + + @Test + void fractionalPlatformWindowIsExpandedThenStrictlyFiltered() throws Exception { + Fixture f = new Fixture(); + Instant start = START.plusMillis(500); + Instant end = START.plusSeconds(3600).plusMillis(250); + f.pull(start, end, 0L); + assertEquals(START.getEpochSecond(), f.body.get().path("time_from").asLong()); + assertEquals( + START.plusSeconds(3600).getEpochSecond(), + f.body.get().path("time_to").asLong()); + + ((ObjectNode) f.items().get(0)).put("time", START.getEpochSecond()); + PullPage filtered = f.pull(start, end, 0L); + assertEquals(1, filtered.getEvents().size()); + assertEquals(2, filtered.getSourceRecordCount()); + } + + @Test + void vendorAndPaginationSchemaFailuresFailClosed() throws Exception { + Fixture vendor = new Fixture(); + vendor.root.put("response_code", -1).put("verbose_msg", SECRET); + AdapterException vendorFailure = assertThrows(AdapterException.class, () -> vendor.pull(0L)); + assertEquals("VENDOR_RESPONSE_ERROR", vendorFailure.getCode()); + assertFalse(vendorFailure.toString().contains(SECRET)); + + Fixture total = new Fixture(); + total.root.with("data").put("total", 3); + PullPage capacityProbe = total.pull(0L); + assertEquals(3L, capacityProbe.getTotal()); + assertEquals(2, capacityProbe.getSourceRecordCount()); + + Fixture inconsistent = new Fixture(); + inconsistent.root.with("data").put("total", 1); + assertEquals( + "PAGINATION_INCONSISTENT", + assertThrows(AdapterException.class, () -> inconsistent.pull(0L)).getCode()); + + Fixture duplicate = new Fixture(); + ((ObjectNode) duplicate.items().get(1)).put("id", "log-1"); + assertEquals( + "PAGINATION_DUPLICATE_DOCUMENT", + assertThrows(AdapterException.class, () -> duplicate.pull(0L)).getCode()); + + Fixture missing = new Fixture(); + missing.root.with("data").remove("total"); + assertEquals( + "RESPONSE_SCHEMA_ERROR", + assertThrows(AdapterException.class, () -> missing.pull(0L)).getCode()); + } + + @Test + void responseBoundsTimeoutsAndHttpErrorsAreClassifiedWithoutBodyLeaks() throws Exception { + Fixture oversized = new Fixture(); + oversized.config.setMaxResponseBytes(1024L); + ((ObjectNode) oversized.items().get(0)).put("large", repeat('x', 2000)); + assertEquals( + "RESPONSE_TOO_LARGE", + assertThrows(AdapterException.class, () -> oversized.pull(0L)).getCode()); + + Fixture timeout = new Fixture(); + when(timeout.http.execute(any(HttpUriRequest.class))) + .thenThrow(new SocketTimeoutException(SECRET)); + AdapterException timeoutFailure = + assertThrows(AdapterException.class, () -> timeout.pull(0L)); + assertEquals("HTTP_TIMEOUT", timeoutFailure.getCode()); + assertFalse(timeoutFailure.toString().contains(SECRET)); + + Fixture forbidden = new Fixture(); + forbidden.status = 403; + forbidden.rawOverride = SECRET; + AdapterException httpFailure = + assertThrows(AdapterException.class, () -> forbidden.pull(0L)); + assertEquals("AUTH_HTTP_403", httpFailure.getCode()); + assertFalse(httpFailure.toString().contains(SECRET)); + } + + @Test + void configAndCredentialSchemasRejectCrossVendorOrUnsafeShapesBeforeHttp() throws Exception { + Fixture f = new Fixture(); + f.config.setBaseUrl("https://192.0.2.10/api"); + assertEquals( + "CONFIG_SCHEMA_ERROR", + assertThrows(AdapterException.class, () -> f.pull(0L)).getCode()); + verifyNoInteractions(f.http); + + Fixture unpaged = new Fixture(); + unpaged.config.setMaxPagesPerRun(2); + assertEquals( + "CONFIG_SCHEMA_ERROR", + assertThrows(AdapterException.class, () -> unpaged.pull(0L)).getCode()); + verifyNoInteractions(unpaged.http); + + Fixture incident = new Fixture(); + incident.config.setStreamKey("WEIBU_INCIDENT"); + assertEquals( + "STREAM_NOT_SUPPORTED", + assertThrows(AdapterException.class, () -> incident.pull(0L)).getCode()); + verifyNoInteractions(incident.http); + + CredentialSchemas.validate(WeibuApiAdapter.TYPE, API_KEY, SECRET, null); + assertThrows( + CredentialException.class, + () -> CredentialSchemas.validate(WeibuApiAdapter.TYPE, API_KEY, SECRET, "token")); + assertThrows( + CredentialException.class, + () -> CredentialSchemas.validate(WeibuApiAdapter.TYPE, null, SECRET, null)); + } + + private static String repeat(char value, int count) { + char[] chars = new char[count]; + java.util.Arrays.fill(chars, value); + return new String(chars); + } + + private static Map query(HttpPost request) { + Map result = new LinkedHashMap<>(); + List pairs = + URLEncodedUtils.parse(request.getURI(), StandardCharsets.UTF_8); + for (NameValuePair pair : pairs) result.put(pair.getName(), pair.getValue()); + return result; + } + + private static DeviceCollectApiConfig config() { + DeviceCollectApiConfig c = new DeviceCollectApiConfig(); + c.setCollectTaskId(22); + c.setDeviceId(9); + c.setAdapterType(WeibuApiAdapter.TYPE); + c.setAdapterVersion(1); + c.setStreamKey(WeibuApiAdapter.LOG); + c.setBaseUrl("https://192.0.2.10"); + c.setCredentialRef("weibu-fixture"); + c.setPageSize(2); + c.setPollIntervalSeconds(30); + c.setMaxPagesPerRun(1); + c.setConnectTimeoutMs(1000); + c.setRequestTimeoutMs(3000); + c.setWindowSizeSeconds(3600); + c.setSafetyDelaySeconds(120); + c.setLookbackSeconds(300); + c.setMaxResponseBytes(1024 * 1024L); + c.setExtraConfigJson("{}"); + return c; + } + + private static ObjectNode response() { + ObjectNode root = JSON.createObjectNode(); + root.put("response_code", 0); + ObjectNode data = root.putObject("data"); + data.put("total", 2); + ArrayNode items = data.putArray("data"); + items.addObject() + .put("id", "log-1") + .put("machine", "192.0.2.10") + .put("time", START.plusSeconds(100).getEpochSecond()) + .put("direction", "in") + .putObject("threat") + .put("level", "attack") + .put("name", "synthetic attack one"); + items.addObject() + .put("id", "log-2") + .put("machine", "192.0.2.11") + .put("time", START.plusSeconds(200).getEpochSecond()) + .put("direction", "lateral") + .putObject("threat") + .put("level", "risk") + .put("name", "synthetic risk two"); + return root; + } + + private static CloseableHttpResponse response(int status, HttpEntity body) { + CloseableHttpResponse result = mock(CloseableHttpResponse.class); + StatusLine statusLine = mock(StatusLine.class); + when(statusLine.getStatusCode()).thenReturn(status); + when(result.getStatusLine()).thenReturn(statusLine); + when(result.getEntity()).thenReturn(body); + return result; + } + + private static final class Fixture { + private final CloseableHttpClient http = mock(CloseableHttpClient.class); + private final AtomicReference request = new AtomicReference<>(); + private final AtomicReference body = new AtomicReference<>(); + private final ObjectNode root = response(); + private DeviceCollectApiConfig config = config(); + private int status = 200; + private boolean unknownLength; + private String rawOverride; + private final WeibuApiAdapter adapter; + + private Fixture() throws Exception { + ApiPullProperties properties = new ApiPullProperties(); + properties.setAllowedHosts("192.0.2.10"); + CredentialProvider provider = + (ref, type, device) -> { + Map values = new LinkedHashMap<>(); + values.put("app_key", new SecretValue(API_KEY.toCharArray())); + values.put("app_secret", new SecretValue(SECRET.toCharArray())); + return new CredentialSnapshot(1, "fixture-v1", type, values); + }; + adapter = + new WeibuApiAdapter( + provider, new SafeBaseUrlValidator(properties), JSON, http); + when(http.execute(any(HttpUriRequest.class))) + .thenAnswer( + invocation -> { + HttpPost post = invocation.getArgument(0); + request.set(post); + body.set(JSON.readTree(EntityUtils.toString(post.getEntity(), "UTF-8"))); + String text = rawOverride == null ? root.toString() : rawOverride; + HttpEntity entity; + if (unknownLength) { + entity = mock(HttpEntity.class); + when(entity.getContentLength()).thenReturn(-1L); + when(entity.getContent()) + .thenReturn( + new ByteArrayInputStream(text.getBytes(StandardCharsets.UTF_8))); + } else { + entity = new StringEntity(text, "UTF-8"); + } + return response(status, entity); + }); + } + + private ArrayNode items() { + return (ArrayNode) root.path("data").path("data"); + } + + private PullPage pull(long offset) { + return pull(START, END, offset); + } + + private PullPage pull(Instant start, Instant end, long offset) { + return adapter.pull(new PullRequest(config, start, end, 2, offset)); + } + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/credential/DatabaseCredentialProviderTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/credential/DatabaseCredentialProviderTest.java index 809cfd9..8bae325 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/credential/DatabaseCredentialProviderTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/credential/DatabaseCredentialProviderTest.java @@ -39,6 +39,23 @@ class DatabaseCredentialProviderTest { assertEquals("CREDENTIAL_NOT_FOUND", error.getCode()); } + @Test + void weibuUsesTheExistingAppKeyAndSecretCredentialColumns() { + DeviceApiCredentialMapper mapper = mock(DeviceApiCredentialMapper.class); + DeviceApiCredential record = credential(); + record.setCredentialRef("weibu-v1"); + record.setDeviceId(9); + record.setAdapterType("WEIBU_V1"); + when(mapper.selectForLoad("weibu-v1", 9, "WEIBU_V1")).thenReturn(record); + + DatabaseCredentialProvider provider = new DatabaseCredentialProvider(mapper); + try (CredentialSnapshot snapshot = provider.load("weibu-v1", "WEIBU_V1", 9)) { + assertArrayEquals("app-key".toCharArray(), snapshot.require("app_key").copy()); + assertEquals("[REDACTED]", snapshot.require("app_secret").toString()); + assertThrows(CredentialException.class, () -> snapshot.require("api_token")); + } + } + private DeviceApiCredential credential() { DeviceApiCredential record = new DeviceApiCredential(); record.setCredentialRef("ndr-v1"); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/security/SafeBaseUrlValidatorTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/security/SafeBaseUrlValidatorTest.java new file mode 100644 index 0000000..7faeec8 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/security/SafeBaseUrlValidatorTest.java @@ -0,0 +1,63 @@ +package com.haobang.apipull.security; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.haobang.apipull.ApiPullProperties; +import com.haobang.apipull.adapter.AdapterException; +import org.junit.jupiter.api.Test; + +class SafeBaseUrlValidatorTest { + @Test + void disabledAllowlistAcceptsNewHttpsDeviceWithoutConfigurationChange() { + ApiPullProperties properties = new ApiPullProperties(); + properties.setHostAllowlistEnabled(false); + properties.setAllowedHosts(""); + + java.net.URI accepted = + new SafeBaseUrlValidator(properties).validate("https://192.0.2.20/"); + assertEquals("https", accepted.getScheme()); + assertEquals("192.0.2.20", accepted.getHost()); + } + + @Test + void enabledAllowlistStillUsesExactHostMatching() { + ApiPullProperties properties = new ApiPullProperties(); + properties.setHostAllowlistEnabled(true); + properties.setAllowedHosts("192.0.2.10"); + SafeBaseUrlValidator validator = new SafeBaseUrlValidator(properties); + + assertEquals( + "https://192.0.2.10", validator.validate("https://192.0.2.10").toString()); + AdapterException failure = + assertThrows( + AdapterException.class, () -> validator.validate("https://192.0.2.20")); + assertEquals("SSRF_REJECTED", failure.getCode()); + } + + @Test + void disabledAllowlistDoesNotRelaxUrlShapeValidation() { + ApiPullProperties properties = new ApiPullProperties(); + properties.setHostAllowlistEnabled(false); + SafeBaseUrlValidator validator = new SafeBaseUrlValidator(properties); + + assertEquals( + "SSRF_REJECTED", + assertThrows( + AdapterException.class, + () -> validator.validate("http://192.0.2.20/path")) + .getCode()); + assertEquals( + "SSRF_REJECTED", + assertThrows( + AdapterException.class, + () -> validator.validate("https://user@192.0.2.20/path")) + .getCode()); + assertEquals( + "SSRF_REJECTED", + assertThrows( + AdapterException.class, + () -> validator.validate("https://192.0.2.20/../admin")) + .getCode()); + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/DeviceContextResolverTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/DeviceContextResolverTest.java index 1f63cbc..cf7bb95 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/DeviceContextResolverTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/DeviceContextResolverTest.java @@ -1,6 +1,7 @@ package com.haobang.dispatch; import com.common.entity.DeviceDevice; +import com.common.entity.DeviceCollectionMode; import com.common.service.DeviceDeviceService; import com.common.service.DeviceUnknownService; import com.haobang.config.AppProperties; @@ -16,6 +17,8 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; class DeviceContextResolverTest { + private static final String IP = "192.168.4.19"; + @Test void repeatedSourceIpQueriesDatabaseOnlyOnceWithinTtl() { DeviceDeviceService service = mock(DeviceDeviceService.class); @@ -83,4 +86,90 @@ class DeviceContextResolverTest { assertTrue(resolver.isIngressSnapshotReady()); assertEquals(1, resolver.getIngressSnapshotSize()); } + + @Test + void apiOnlyIsInvalidForIngressButAllowedByDeviceId() { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceDevice device = device(DeviceCollectionMode.API_ONLY); + when(service.getActiveDevices()).thenReturn(Collections.singletonList(device)); + when(service.getByIdSafely(126)).thenReturn(device); + DeviceContextResolver resolver = resolver(service); + + resolver.refreshIngressSnapshot(); + + assertEquals(DeviceContextResolver.IngressStatus.INVALID, + resolver.resolveIngress(IP, "UDP")); + assertTrue(resolver.resolveDeviceId(126).isRegistered()); + } + + @Test + void syslogOnlyDeniesApiWithStableResolutionType() { + DeviceDeviceService service = mock(DeviceDeviceService.class); + when(service.getByIdSafely(126)).thenReturn(device(DeviceCollectionMode.SYSLOG_ONLY)); + + DeviceContextResolver.Resolution result = resolver(service).resolveDeviceId(126); + + assertEquals(DeviceContextResolver.Type.COLLECTION_MODE_DENIED, result.getType()); + assertTrue(result.getReason().contains("API_PULL")); + } + + @Test + void bothAllowsSyslogAndApi() { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceDevice device = device(DeviceCollectionMode.BOTH); + when(service.getByIpSafely(IP)).thenReturn(Collections.singletonList(device)); + when(service.getByIdSafely(126)).thenReturn(device); + DeviceContextResolver resolver = resolver(service); + + assertTrue(resolver.resolve(IP, "TCP").isRegistered()); + assertTrue(resolver.resolveDeviceId(126).isRegistered()); + } + + @Test + void nullCollectionModeFromLegacyCacheBehavesAsBoth() { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceDevice device = device(null); + when(service.getByIpSafely(IP)).thenReturn(Collections.singletonList(device)); + when(service.getByIdSafely(126)).thenReturn(device); + DeviceContextResolver resolver = resolver(service); + + assertEquals(DeviceCollectionMode.BOTH, device.getCollectionMode()); + assertTrue(resolver.resolve(IP, "TCP").isRegistered()); + assertTrue(resolver.resolveDeviceId(126).isRegistered()); + } + + @Test + void invalidateDeviceClearsBothIpAndIdLocalEntries() { + DeviceDeviceService service = mock(DeviceDeviceService.class); + DeviceDevice device = device(DeviceCollectionMode.BOTH); + when(service.getByIpSafely(IP)).thenReturn(Collections.singletonList(device)); + when(service.getByIdSafely(126)).thenReturn(device); + DeviceContextResolver resolver = resolver(service); + resolver.resolve(IP, "TCP"); + resolver.resolveDeviceId(126); + + resolver.invalidateDevice(126, IP); + resolver.resolve(IP, "TCP"); + resolver.resolveDeviceId(126); + + verify(service, times(2)).getByIpSafely(IP); + verify(service, times(2)).getByIdSafely(126); + } + + private DeviceContextResolver resolver(DeviceDeviceService service) { + AppProperties properties = new AppProperties(); + properties.setAppServiceDeviceCollectId(1); + properties.setSyslogDispatchTerminalRecordEnabled(false); + properties.setSyslogDeviceContextCacheTtlMs(60000L); + return new DeviceContextResolver(service, mock(DeviceUnknownService.class), properties); + } + + private DeviceDevice device(DeviceCollectionMode mode) { + DeviceDevice device = new DeviceDevice(); + device.setId(126); + device.setIp(IP); + device.setDeviceCollectId(1); + device.setCollectionMode(mode); + return device; + } } diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java index a712119..3844c6b 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java @@ -1,6 +1,9 @@ package com.haobang.dispatch; import com.common.entity.DeviceDevice; +import com.common.entity.DeviceCollectionMode; +import com.common.service.DeviceDeviceService; +import com.common.service.DeviceUnknownService; import com.fasterxml.jackson.databind.ObjectMapper; import com.haobang.config.AppProperties; import com.haobang.diagnostics.SyslogIpMetricsRegistry; @@ -19,6 +22,7 @@ import java.time.LocalDateTime; import java.time.Instant; import java.nio.charset.StandardCharsets; import java.util.concurrent.CompletableFuture; +import java.util.Collections; import java.util.stream.Stream; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -58,6 +62,39 @@ class SyslogWalDispatcherTest { verify(producer).messagePush(eq(device),eq("{}"),eq("20260831100102123"),eq("20260831100058127"),eq("20260831-api"),eq(event.getPointTimeNs())); } + @Test + void syslogOnlyApiRecordAlreadyInWalIsStoppedByDispatcher() throws Exception { + AppProperties properties = properties(tempDirectory); + properties.setAppServiceDeviceCollectId(1); + properties.setSyslogDispatchTerminalRecordEnabled(true); + wal = new SyslogWalService(properties, new ObjectMapper()); + wal.initialize(); + IngressEvent event = new IngressEvent( + "20260831-denied-api", Instant.parse("2026-08-31T02:01:02.123Z"), + Instant.parse("2026-08-31T02:00:58.127Z"), IngressType.API_PULL, "HTTPS", 40, 2001, + "WEIBU_V1", "incident:1", "weibu.example", 443, "JSON", + StandardCharsets.UTF_8, "{}", null); + wal.tryAppendUnified(wal.prepareIngressEvent(event)); + waitUntil(() -> wal.getAppendedCount() == 1L); + DeviceDevice device = new DeviceDevice(); + device.setId(40); + device.setDeviceCollectId(1); + device.setCollectionMode(DeviceCollectionMode.SYSLOG_ONLY); + DeviceDeviceService devices = mock(DeviceDeviceService.class); + when(devices.getByIdSafely(40)).thenReturn(device); + DeviceContextResolver resolver = new DeviceContextResolver( + devices, mock(DeviceUnknownService.class), properties); + kafkaProducer producer = mock(kafkaProducer.class); + + dispatcher = new SyslogWalDispatcher(wal, resolver, producer, properties); + dispatcher.start(); + + waitUntil(() -> wal.getCheckpointSequence() == 1L); + assertEquals(1L, dispatcher.getTerminalFiltered()); + assertEquals(0L, dispatcher.getKafkaAcked()); + verify(producer, never()).messagePush(any(), anyString(), anyString(), anyString(), any()); + } + @Test void outOfOrderKafkaAckAdvancesOnlyContiguousCheckpoint() throws Exception { AppProperties properties = properties(tempDirectory); @@ -128,6 +165,36 @@ class SyslogWalDispatcherTest { verify(producer, never()).messagePush(any(), anyString(), anyString(), anyString(), any()); } + @Test + void apiOnlySyslogAlreadyInWalIsStoppedByDispatcher() throws Exception { + AppProperties properties = properties(tempDirectory); + properties.setAppServiceDeviceCollectId(1); + properties.setSyslogDispatchTerminalRecordEnabled(true); + wal = new SyslogWalService(properties, new ObjectMapper()); + wal.initialize(); + wal.tryAppend(event("20260811-api-only-syslog", "must-not-reach-kafka")); + waitUntil(() -> wal.getAppendedCount() == 1L); + DeviceDevice device = new DeviceDevice(); + device.setId(40); + device.setIp("192.168.4.19"); + device.setDeviceCollectId(1); + device.setCollectionMode(DeviceCollectionMode.API_ONLY); + DeviceDeviceService devices = mock(DeviceDeviceService.class); + when(devices.getByIpSafely("192.168.4.19")) + .thenReturn(Collections.singletonList(device)); + DeviceContextResolver resolver = new DeviceContextResolver( + devices, mock(DeviceUnknownService.class), properties); + kafkaProducer producer = mock(kafkaProducer.class); + + dispatcher = new SyslogWalDispatcher(wal, resolver, producer, properties); + dispatcher.start(); + + waitUntil(() -> wal.getCheckpointSequence() == 1L); + assertEquals(1L, dispatcher.getTerminalFiltered()); + assertEquals(0L, dispatcher.getKafkaAcked()); + verify(producer, never()).messagePush(any(), anyString(), anyString(), anyString(), any()); + } + @Test void terminalDeviceStateIsDroppedWithoutFileWhenRecordingDisabled() throws Exception { AppProperties properties = properties(tempDirectory); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java index 78933d6..0a732d9 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java @@ -1,5 +1,9 @@ package com.netty; +import com.common.entity.DeviceCollectionMode; +import com.common.entity.DeviceDevice; +import com.common.service.DeviceDeviceService; +import com.common.service.DeviceUnknownService; import com.haobang.dispatch.DeviceContextResolver; import com.haobang.config.AppProperties; import com.haobang.diagnostics.SyslogIpMetricsRegistry; @@ -10,6 +14,7 @@ import org.junit.jupiter.api.Test; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.util.Collections; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -73,18 +78,31 @@ class SyslogMessageHandlerBackpressureTest { } @Test - void invalidDeviceIsDroppedBeforeWalInPerformanceMode() { + void apiOnlyDeviceIsCountedAndDroppedBeforeWalInPerformanceMode() { SyslogWalService wal = mock(SyslogWalService.class); - DeviceContextResolver resolver = mock(DeviceContextResolver.class); - when(resolver.resolveIngress(anyString(), anyString())) - .thenReturn(DeviceContextResolver.IngressStatus.INVALID); + DeviceDeviceService devices = mock(DeviceDeviceService.class); + DeviceDevice device = new DeviceDevice(); + device.setId(40); + device.setIp("embedded"); + device.setDeviceCollectId(1); + device.setCollectionMode(DeviceCollectionMode.API_ONLY); + when(devices.getActiveDevices()).thenReturn(Collections.singletonList(device)); + AppProperties properties = new AppProperties(); + properties.setAppServiceDeviceCollectId(1); + properties.setSyslogDispatchTerminalRecordEnabled(false); + DeviceContextResolver resolver = new DeviceContextResolver( + devices, mock(DeviceUnknownService.class), properties); + resolver.refreshIngressSnapshot(); + long networkBefore = SyslogMessageHandler.getNetworkReceivedCount(); + long invalidBefore = SyslogMessageHandler.getInvalidDeviceDroppedCount(); EmbeddedChannel channel = new EmbeddedChannel( new SyslogMessageHandler(StandardCharsets.UTF_8, wal, resolver, true)); channel.connect(new InetSocketAddress("192.168.4.19", 5514)); channel.writeInbound("<134>Aug 11 20:00:00 invalid device event"); - verify(resolver).resolveIngress("embedded", "TCP"); + assertEquals(networkBefore + 1L, SyslogMessageHandler.getNetworkReceivedCount()); + assertEquals(invalidBefore + 1L, SyslogMessageHandler.getInvalidDeviceDroppedCount()); verify(wal, never()).prepareForWal(any()); verify(wal, never()).tryAppend(any()); channel.close(); diff --git a/haobang-security-dm/upgrade-doc/README.md b/haobang-security-dm/upgrade-doc/README.md index 45f3eee..78cb7f0 100644 --- a/haobang-security-dm/upgrade-doc/README.md +++ b/haobang-security-dm/upgrade-doc/README.md @@ -9,9 +9,10 @@ | `v3/` | 日统计异步投影，以及大日志独立 Kafka 通道、字节背压和存储截断升级资料 | | `v4/` | Kafka 积压分级采样、告警优先落库和 InfluxDB 非阻塞旁路升级资料 | | `v5/` | 按设备短时采集未截断原始日志，并将异常样本优先落盘的现场诊断工具 | +| `v6/` | 设备级采集模式：为 `device_device` 增加 `collection_mode`，支持 SYSLOG_ONLY、API_ONLY、BOTH | 新环境按 v1、v2、v3、v4、v5 的顺序执行；已经完成 v4 的现场只需升级应用并执行 v5。已经执行过早期 -v3 日统计升级的现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，再执行 v4、v5；不要重复执行 +v3 日统计升级的现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，再执行 v4、v5、v6；不要重复执行 数据库升级脚本。 数据库脚本不要重复执行。执行过程中有报错时，先确认已经落库的对象，再决定从哪里继续。 @@ -20,3 +21,4 @@ v2 的执行步骤见 [v2/README.md](./v2/README.md)。 v3 的执行步骤见 [v3/README.md](./v3/README.md)。 v4 的执行步骤见 [v4/README.md](./v4/README.md)。 v5 的执行步骤见 [v5/README.md](./v5/README.md)。 +v6 的执行步骤见 [v6/README.md](./v6/README.md)。 diff --git a/haobang-security-dm/upgrade-doc/v6/README.md b/haobang-security-dm/upgrade-doc/v6/README.md new file mode 100644 index 0000000..16cb8f1 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/README.md @@ -0,0 +1,59 @@ +# v6 设备级采集模式增量升级说明 + +## 默认采集总开关 + +新版默认同时开启 Syslog 服务和设备 API 拉取调度： + +```text +SYSLOG_SERVER_ENABLED=true +DEVICE_API_ENABLED=true +``` + +`DEVICE_DEVICE.COLLECTION_MODE` 默认 `BOTH`，因此既有设备升级后同时允许 Syslog 和 API。具体 API +任务仍需完成校验并通过 `resume` 启用；全局开关为 true 不会自动启动暂停任务。 + +## 必须执行的数据库脚本 + +新版 `syslog-serve` 会直接查询 `device_device.collection_mode`。因此必须在启动新版 JAR 前，由 DBA 使用 `syslog-serve` 实际连接的 Schema 按顺序执行： + +```text +database/01-device-collection-mode-precheck.sql +database/02-device-collection-mode-upgrade.sql +database/03-device-collection-mode-verify.sql +``` + +执行规则： + +1. 先备份 `DEVICE_DEVICE`，保存预检结果。 +2. `01` 中字段和约束查询首次升级时必须返回 0 行。 +3. `02` 只能成功执行一次；中途报错时不要整份重跑，应先检查已落库对象。 +4. `03` 必须确认字段为 `VARCHAR2(16) DEFAULT 'BOTH' NOT NULL`、约束已启用、非法数据数为 0。 +5. `03` 验证完成后即可启动新版 JAR。先配置下游微步 API JSON 泛化规则，再写入微步凭据、创建并验证 `WEIBU_LOG` 新任务；确认规则已关联设备后才能恢复任务并设置具体设备模式。 + +默认 `BOTH` 用于兼容现有设备；升级 SQL 不硬编码任何设备 ID。 + +本次 v6 强制数据库结构变更只有 `DEVICE_DEVICE.COLLECTION_MODE`。微步 API 复用现有 +`DEVICE_COLLECT_API_*` 基础表，不在本次升级中新增或修改这些表。新版微步适配器只支持 +`WEIBU_LOG`（`/api/v1/log/searchBySql`）；开发阶段的 `WEIBU_INCIDENT` 从未作为现场版本发布，因此不需要现场数据迁移脚本。 + +## 现场设置示例 + +```bash +docker exec syslog-serve-dm curl -fsS -X PUT \ + 'http://127.0.0.1:8189/syslogserve/api/devices/40/collection-mode' \ + -H 'Content-Type: application/json' \ + -d '{"collectionMode":"API_ONLY"}' + +docker exec syslog-serve-dm curl -fsS \ + 'http://127.0.0.1:8189/syslogserve/api/devices/40' +``` + +GET 响应必须包含 `"collectionMode":"API_ONLY"`。设置后，设备 40 的微步 API 拉取任务继续运行，IP `10.56.171.164` 的 TCP/UDP Syslog 不得进入 WAL。 + +多实例部署时，收到 PUT 请求的实例立即清理三层缓存并刷新入站快照；其他实例最迟在现有设备快照 TTL 后生效。 + +微步凭据与任务、设备 `API_ONLY`、缓存清理、验收和回退已合并为一份现场手册： + +- [api/README.md](./api/README.md) + +按现场要求，[api/weibu-40-credential.json](./api/weibu-40-credential.json) 已填写设备 40 的真实微步 ApiKey/Secret；[api/weibu-40-task.json](./api/weibu-40-task.json) 已填写设备、地址和拉取参数，首次采集时间由现场命令生成。凭据文件包含明文生产秘密，必须限制整个 v6 目录的权限和分发范围；禁止把文件内容复制到工单、聊天或执行日志。 diff --git a/haobang-security-dm/upgrade-doc/v6/api/README.md b/haobang-security-dm/upgrade-doc/v6/api/README.md new file mode 100644 index 0000000..f6d8d90 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/api/README.md @@ -0,0 +1,149 @@ +# v6 API 现场操作 + +API 相关升级只看本页。示例设备：`deviceId=40`、`IP=10.56.171.164`、微步 TDP。 +生产环境不映射管理端口，所有 API 命令都通过容器执行： + +- `syslog-serve-dm`：`8189` +- `syslog-consumer-rule-dm`：`8289` + +## 1. 发布顺序 + +1. 先按 `upgrade-doc/v6/README.md` 执行 3 个数据库脚本。 +2. 确认启动配置： + + ```text + SYSLOG_SERVER_ENABLED=true + DEVICE_API_ENABLED=true + DEVICE_API_TLS_VERIFY_ENABLED=false + DEVICE_API_HOST_ALLOWLIST_ENABLED=false + ``` + +3. 更换并启动新 JAR。新版 JAR 不能在数据库字段升级前启动。 +4. 检查启动日志： + + ```bash + docker logs syslog-serve-dm 2>&1 | grep '\[SYSLOG_BUILD_READY\]' | tail -n 1 + ``` + + 必须看到 `apiPullEnabled=true`。内网默认 `apiTlsVerifyEnabled=false`。 + +## 2. 配置微步 API 任务 + +真实 Key/Secret 已保存在 `weibu-40-credential.json`，该文件必须按生产密钥管理。 + +```bash +chmod 600 upgrade-doc/v6/api/weibu-40-credential.json + +docker exec -i syslog-serve-dm curl -fsS -X POST \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-credentials' \ + -H 'Content-Type: application/json' \ + --data-binary '@-' \ + < upgrade-doc/v6/api/weibu-40-credential.json +``` + +如果凭据已存在，不要重复创建。任务已存在时也直接复用。首次接入时，v6 目录已提供设备 40 的完整任务 JSON；执行前只把首次采集时间设为服务器当前时间减 2 分钟： + +```bash +INITIAL_START_TIME=$(date -d '2 minutes ago' --iso-8601=seconds) +sed "s|__INITIAL_START_TIME__|${INITIAL_START_TIME}|" \ + upgrade-doc/v6/api/weibu-40-task.json \ + > /tmp/weibu-40-task.json +``` + +执行前检查生成结果： + +```bash +cat /tmp/weibu-40-task.json +``` + +必须确认 `deviceId=40`、`baseUrl=https://10.56.171.164`，且 `initialStartTime` 已经是真实时间，不再包含 `__INITIAL_START_TIME__`。 + +```bash +docker exec -i syslog-serve-dm curl -fsS -X POST \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks' \ + -H 'Content-Type: application/json' \ + --data-binary '@-' \ + < /tmp/weibu-40-task.json +``` + +记录返回的 `taskId` 和 `configVersion`。任务创建后默认暂停，先验证： + +```bash +docker exec syslog-serve-dm curl -fsS -X POST \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks//validate' \ + -H 'Content-Type: application/json' \ + -d '{"configVersion":}' +``` + +所有检查为 `PASS` 后，先在管理系统中给设备 40 关联能解析微步单条 JSON 的泛化规则，然后恢复任务： + +```bash +docker exec syslog-serve-dm curl -fsS -X POST \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks//resume' \ + -H 'Content-Type: application/json' \ + -d '{"configVersion":}' +``` + +## 3. 将设备 40 设为仅 API 采集 + +禁止直接改数据库，必须调用 API，让三层设备缓存同时失效。 + +```bash +docker exec syslog-serve-dm curl -fsS -X PUT \ + 'http://127.0.0.1:8189/syslogserve/api/devices/40/collection-mode' \ + -H 'Content-Type: application/json' \ + -d '{"collectionMode":"API_ONLY"}' + +docker exec syslog-serve-dm curl -fsS \ + 'http://127.0.0.1:8189/syslogserve/api/devices/40' +``` + +GET 必须返回 `collectionMode=API_ONLY`。此后 `10.56.171.164` 的 TCP/UDP Syslog 被丢弃，API 任务继续运行。 + +## 4. 验收 + +```bash +docker exec syslog-serve-dm curl -fsS \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks//runtime' +docker exec syslog-serve-dm curl -fsS \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks//runs?page=1&pageSize=5' +``` + +必须确认： + +- `health=HEALTHY`，最新 `SCHEDULED` 为 `SUCCESS`，HTTP 状态为 200。 +- checkpoint 持续推进；设备当时无日志时 `recordsFetched=0` 也是正常成功。 +- Kafka 中每条消息是一条微步 JSON，下游按设备 40 关联的泛化规则标准化。 +- 来自 `10.56.171.164` 的 Syslog 不进入 WAL/Kafka，其他默认 `BOTH` 设备不受影响。 + +## 5. 缓存清理 + +```bash +# syslog-serve：指定设备 / 全部设备 +docker exec syslog-serve-dm curl -fsS -X DELETE \ + 'http://127.0.0.1:8189/syslogserve/api/devices/40/cache' +docker exec syslog-serve-dm curl -fsS -X DELETE \ + 'http://127.0.0.1:8189/syslogserve/api/devices/cache' + +# syslog-consumer-rule：指定设备规则 / 全部规则 +docker exec syslog-consumer-rule-dm curl -fsS -X DELETE \ + 'http://127.0.0.1:8289/xdrrule/api/syslog/normalization/cache/devices/40' +docker exec syslog-consumer-rule-dm curl -fsS -X DELETE \ + 'http://127.0.0.1:8289/xdrrule/api/syslog/normalization/cache' +``` + +规则缓存清理会通知所有正常订阅 Redis 消息的 `syslog-consumer-rule` 实例。不要使用旧的 `DELETE /api/cache/all`。 + +## 6. 快速回退 + +1. 暂停微步 API 任务。 +2. 将设备改回 `BOTH`： + + ```bash + docker exec syslog-serve-dm curl -fsS -X PUT \ + 'http://127.0.0.1:8189/syslogserve/api/devices/40/collection-mode' \ + -H 'Content-Type: application/json' \ + -d '{"collectionMode":"BOTH"}' + ``` + +3. 恢复原泛化规则关联和旧 JAR。不删除 `COLLECTION_MODE` 字段，不删除任务、凭据或 checkpoint。 diff --git a/haobang-security-dm/upgrade-doc/v6/api/weibu-40-credential.json b/haobang-security-dm/upgrade-doc/v6/api/weibu-40-credential.json new file mode 100644 index 0000000..d41563a --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/api/weibu-40-credential.json @@ -0,0 +1,8 @@ +{ + "credentialRef": "weibu-40", + "deviceId": 40, + "adapterType": "WEIBU_V1", + "credentialVersion": "v1", + "appKey": "57d8fb2b556a471ea36b8b4441fa1c0c", + "appSecret": "9d967876f51d4e16bbf8bde069d12a9b" +} diff --git a/haobang-security-dm/upgrade-doc/v6/api/weibu-40-task.json b/haobang-security-dm/upgrade-doc/v6/api/weibu-40-task.json new file mode 100644 index 0000000..4a9f416 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/api/weibu-40-task.json @@ -0,0 +1,20 @@ +{ + "deviceId": 40, + "taskName": "微步逐条日志拉取", + "adapterType": "WEIBU_V1", + "streamKey": "WEIBU_LOG", + "baseUrl": "https://10.56.171.164", + "credentialRef": "weibu-40", + "pollIntervalSeconds": 30, + "pageSize": 1000, + "maxPagesPerRun": 1, + "schedulerPriority": 0, + "connectTimeoutMs": 5000, + "requestTimeoutMs": 30000, + "windowSizeSeconds": 300, + "safetyDelaySeconds": 120, + "lookbackSeconds": 300, + "maxResponseBytes": 16777216, + "initialStartTime": "__INITIAL_START_TIME__", + "extraConfig": {} +} diff --git a/haobang-security-dm/upgrade-doc/v6/database/01-device-collection-mode-precheck.sql b/haobang-security-dm/upgrade-doc/v6/database/01-device-collection-mode-precheck.sql new file mode 100644 index 0000000..6415486 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/database/01-device-collection-mode-precheck.sql @@ -0,0 +1,33 @@ +-- v6 设备级采集模式：达梦数据库升级前检查 +-- 本脚本只读。必须先完成检查，再执行 02-device-collection-mode-upgrade.sql。 +-- 新版 syslog-serve JAR 启动前必须完成字段升级和验证。 + +-- 1. 确认当前用户和 DEVICE_DEVICE 表存在。 +SELECT USER AS CURRENT_SCHEMA FROM DUAL; + +SELECT TABLE_NAME +FROM USER_TABLES +WHERE TABLE_NAME = 'DEVICE_DEVICE'; + +-- 2. 记录升级前设备总数、有效设备数，供升级后核对。 +SELECT COUNT(*) AS DEVICE_TOTAL +FROM DEVICE_DEVICE; + +SELECT COUNT(*) AS ACTIVE_DEVICE_TOTAL +FROM DEVICE_DEVICE +WHERE DELETED_AT IS NULL + AND (DEL_FLAG IS NULL OR DEL_FLAG = '0'); + +-- 3. 目标字段应不存在。首次升级时预期返回 0 行；若已有记录，停止执行升级脚本并先核实。 +SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE, DATA_DEFAULT +FROM USER_TAB_COLUMNS +WHERE TABLE_NAME = 'DEVICE_DEVICE' + AND COLUMN_NAME = 'COLLECTION_MODE'; + +-- 4. 约束名应未被占用。首次升级时预期返回 0 行。 +SELECT CONSTRAINT_NAME, CONSTRAINT_TYPE, STATUS +FROM USER_CONSTRAINTS +WHERE CONSTRAINT_NAME = 'CK_DEVICE_COLLECTION_MODE'; + +-- 5. 执行 DDL 前必须完成 DEVICE_DEVICE 可恢复备份，并确认没有未结束的长事务或表锁。 +-- 数据库脚本不得写死现场 device_id；具体设备模式在新版 JAR 启动后通过管理 API 设置。 diff --git a/haobang-security-dm/upgrade-doc/v6/database/02-device-collection-mode-upgrade.sql b/haobang-security-dm/upgrade-doc/v6/database/02-device-collection-mode-upgrade.sql new file mode 100644 index 0000000..3f7de93 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/database/02-device-collection-mode-upgrade.sql @@ -0,0 +1,25 @@ +-- v6 设备级采集模式：达梦数据库升级 +-- 强制顺序：停止/保持旧版应用 -> 备份 -> 01 预检 -> 本脚本 -> 03 验证 -> 启动新版 JAR。 +-- 新版 syslog-serve 的所有设备查询都会读取 COLLECTION_MODE，本脚本必须在新版 JAR 启动前执行。 +-- 本脚本不设置任何现场 device_id；所有既有设备统一回填 BOTH，避免误停原有 Syslog 或 API 任务。 + +ALTER TABLE DEVICE_DEVICE + ADD COLLECTION_MODE VARCHAR2(16) DEFAULT 'BOTH'; + +UPDATE DEVICE_DEVICE +SET COLLECTION_MODE = 'BOTH' +WHERE COLLECTION_MODE IS NULL; + +COMMIT; + +ALTER TABLE DEVICE_DEVICE + MODIFY COLLECTION_MODE VARCHAR2(16) DEFAULT 'BOTH' NOT NULL; + +ALTER TABLE DEVICE_DEVICE + ADD CONSTRAINT CK_DEVICE_COLLECTION_MODE + CHECK (COLLECTION_MODE IN ('SYSLOG_ONLY', 'API_ONLY', 'BOTH')); + +COMMENT ON COLUMN DEVICE_DEVICE.COLLECTION_MODE IS + '设备级采集模式：SYSLOG_ONLY、API_ONLY、BOTH；默认 BOTH'; + +COMMIT; diff --git a/haobang-security-dm/upgrade-doc/v6/database/03-device-collection-mode-verify.sql b/haobang-security-dm/upgrade-doc/v6/database/03-device-collection-mode-verify.sql new file mode 100644 index 0000000..509dc2a --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v6/database/03-device-collection-mode-verify.sql @@ -0,0 +1,28 @@ +-- v6 设备级采集模式：达梦数据库升级后验证 +-- 本脚本只读。所有检查通过后才允许启动新版 syslog-serve JAR。 + +-- 1. 预期 1 行：VARCHAR2(16)、NULLABLE=N、DATA_DEFAULT 包含 BOTH。 +SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE, DATA_DEFAULT +FROM USER_TAB_COLUMNS +WHERE TABLE_NAME = 'DEVICE_DEVICE' + AND COLUMN_NAME = 'COLLECTION_MODE'; + +-- 2. 预期 1 行：CONSTRAINT_TYPE=C、STATUS=ENABLED。 +SELECT CONSTRAINT_NAME, CONSTRAINT_TYPE, STATUS +FROM USER_CONSTRAINTS +WHERE TABLE_NAME = 'DEVICE_DEVICE' + AND CONSTRAINT_NAME = 'CK_DEVICE_COLLECTION_MODE'; + +-- 3. 预期 INVALID_MODE_TOTAL=0。 +SELECT COUNT(*) AS INVALID_MODE_TOTAL +FROM DEVICE_DEVICE +WHERE COLLECTION_MODE IS NULL + OR COLLECTION_MODE NOT IN ('SYSLOG_ONLY', 'API_ONLY', 'BOTH'); + +-- 4. 首次升级且尚未调用管理 API 时，预期只有 BOTH，数量与预检 DEVICE_TOTAL 一致。 +SELECT COLLECTION_MODE, COUNT(*) AS DEVICE_TOTAL +FROM DEVICE_DEVICE +GROUP BY COLLECTION_MODE +ORDER BY COLLECTION_MODE; + +-- 验证脚本不创建测试设备，也不绑定任何现场 device_id。