diff --git a/haobang-security-dm/syslog-serve/Dockerfile b/haobang-security-dm/syslog-serve/Dockerfile index 69856a3..f02ca17 100644 --- a/haobang-security-dm/syslog-serve/Dockerfile +++ b/haobang-security-dm/syslog-serve/Dockerfile @@ -1,4 +1,4 @@ -FROM openjdk:8 +FROM eclipse-temurin:8-jre MAINTAINER chenchunan@139.com @@ -15,4 +15,3 @@ EXPOSE 514 CMD java -jar syslog-serve.jar - diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/common/mapper/DeviceCollectApiMapper.java b/haobang-security-dm/syslog-serve/src/main/java/com/common/mapper/DeviceCollectApiMapper.java index f419770..8aef2a3 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/common/mapper/DeviceCollectApiMapper.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/common/mapper/DeviceCollectApiMapper.java @@ -23,7 +23,9 @@ public interface DeviceCollectApiMapper { DeviceCollectApiCheckpoint selectCheckpoint(@Param("taskId") Integer taskId); - List selectDueConfigs(); + /** 只扫描归属于当前采集探针的到期任务，避免不同探针相互领取或推进任务。 */ + List selectDueConfigs( + @Param("deviceCollectId") Integer deviceCollectId); /** 仅在任务暂停且 configVersion 匹配时更新，并原子递增版本。 */ int updateConfigCas(DeviceCollectApiConfig config); @@ -37,7 +39,8 @@ public interface DeviceCollectApiMapper { @Param("taskId") Integer taskId, @Param("configVersion") long configVersion, @Param("owner") String owner, - @Param("leaseSeconds") int leaseSeconds); + @Param("leaseSeconds") int leaseSeconds, + @Param("deviceCollectId") Integer deviceCollectId); /** 只有当前 owner 与 fencing token 同时匹配时才续租。 */ int renewLease( diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java b/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java index 0f76ac6..c313409 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java @@ -11,8 +11,8 @@ import org.springframework.stereotype.Component; @Component public class SyslogBuildStartupLogger { - // 确认 JAR 包含设备级采集模式、API 拉取及全局 TLS 策略。 - static final String CODE_REVISION = "DEVICE_COLLECTION_API_TLS_V20260917_63"; + // 确认 JAR 包含设备级采集模式、API 拉取、探针归属过滤、微步全量查询及诊断接口。 + static final String CODE_REVISION = "WEIBU_DIAGNOSTIC_V20260917_66"; private final Environment environment; public SyslogBuildStartupLogger(Environment environment) { diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java index 13d87b7..e7af348 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java @@ -1,6 +1,7 @@ package com.haobang.apipull; 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; @@ -33,6 +34,7 @@ public class ApiPullScheduler { private final ApiPullRunner runner; private final SyslogWalService wal; private final DeviceContextResolver deviceResolver; + private final AppProperties appProperties; private ExecutorService executor; private Semaphore global; private final ConcurrentHashMap hosts = new ConcurrentHashMap<>(); @@ -43,12 +45,14 @@ public class ApiPullScheduler { DeviceCollectApiMapper mapper, ApiPullRunner runner, SyslogWalService wal, - DeviceContextResolver deviceResolver) { + DeviceContextResolver deviceResolver, + AppProperties appProperties) { this.properties = properties; this.mapper = mapper; this.runner = runner; this.wal = wal; this.deviceResolver = deviceResolver; + this.appProperties = appProperties; } @PostConstruct @@ -78,7 +82,8 @@ public class ApiPullScheduler { @Scheduled(fixedDelayString = "${device.api.scheduler.scan-interval-ms:5000}") public void scan() { if (!properties.isEnabled()) return; - for (DeviceCollectApiConfig c : mapper.selectDueConfigs()) + for (DeviceCollectApiConfig c + : mapper.selectDueConfigs(appProperties.getAppServiceDeviceCollectId())) try { trySubmit(c, "SCHEDULED"); } catch (Throwable failure) { @@ -146,7 +151,8 @@ public class ApiPullScheduler { c.getCollectTaskId(), c.getConfigVersion(), properties.getInstanceId(), - properties.getLeaseSeconds()) + properties.getLeaseSeconds(), + appProperties.getAppServiceDeviceCollectId()) != 1) { releaseLocal(c.getCollectTaskId(), hostPermit); return null; @@ -210,7 +216,11 @@ public class ApiPullScheduler { private boolean requireApiCollectionMode(DeviceCollectApiConfig c, boolean failRequest) { DeviceContextResolver.Resolution resolution = deviceResolver.resolveDeviceId(c.getDeviceId()); if (resolution.isRegistered()) return true; - mapper.scheduleNext(c.getCollectTaskId(), c.getConfigVersion(), c.getPollIntervalSeconds()); + // 非归属节点不能修改任务调度时间；设备归属可能在扫描后、执行前发生变化。 + // 手工触发失败同样不应改变正常调度计划。 + if (!failRequest && resolution.getType() != DeviceContextResolver.Type.NOT_ASSIGNED) { + mapper.scheduleNext(c.getCollectTaskId(), c.getConfigVersion(), c.getPollIntervalSeconds()); + } String reason = resolution.getReason() == null ? "设备不允许 API_PULL" : resolution.getReason(); log.warn("设备 API 拉取已被设备策略跳过: taskId={}, deviceId={}, type={}, reason={}", c.getCollectTaskId(), c.getDeviceId(), resolution.getType(), reason); diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullTaskService.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullTaskService.java index 483ad13..ef033f8 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullTaskService.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullTaskService.java @@ -7,12 +7,14 @@ import com.common.mapper.DeviceCollectApiMapper; import com.common.mapper.DeviceCollectTaskMapper; import com.common.service.DeviceDeviceService; import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import com.haobang.apipull.adapter.AdapterException; import com.haobang.apipull.adapter.AdapterRegistry; import com.haobang.apipull.adapter.ApiPullAdapter; import com.haobang.apipull.adapter.PullRequest; +import com.haobang.apipull.adapter.weibu.WeibuApiAdapter; import com.haobang.apipull.api.ApiPullDtos.*; import com.haobang.apipull.credential.CredentialProvider; import com.haobang.apipull.credential.CredentialSnapshot; @@ -409,6 +411,40 @@ public class ApiPullTaskService { return r; } + /** + * 通过任务既有地址和凭据执行微步原始查询，用于现场比较不同查询及分页字段。 + * 该操作不写运行审计、WAL、Kafka 或 checkpoint。 + */ + public WeibuDiagnosticQueryResponse diagnosticWeibuQuery(Integer taskId, JsonNode requestBody) { + DeviceCollectApiConfig c = requireConfig(taskId); + requireApiAllowedDevice(c.getDeviceId()); + ApiPullAdapter installed = adapters.require(c.getAdapterType()); + if (!(installed instanceof WeibuApiAdapter)) { + throw error( + "ADAPTER_NOT_SUPPORTED", + "该诊断接口仅支持 WEIBU_V1 任务", + HttpStatus.UNPROCESSABLE_ENTITY); + } + WeibuApiAdapter.DiagnosticResult result = + ((WeibuApiAdapter) installed).diagnosticQuery(c, requestBody); + WeibuDiagnosticQueryResponse response = new WeibuDiagnosticQueryResponse(); + response.setTaskId(taskId); + response.setDeviceId(c.getDeviceId()); + response.setConfigVersion(c.getConfigVersion()); + response.setHttpStatus(result.getHttpStatus()); + response.setDurationMs(result.getDurationMs()); + response.setResponseBytes(result.getResponseBytes()); + response.setResponse(result.getResponse()); + log.info( + "微步 API 诊断查询完成: taskId={}, deviceId={}, httpStatus={}, durationMs={}, responseBytes={}", + taskId, + c.getDeviceId(), + result.getHttpStatus(), + result.getDurationMs(), + result.getResponseBytes()); + return response; + } + /** 人工重置已提交窗口。 仅允许暂停且无有效租约的任务，并要求说明原因和 checkpoint 乐观锁版本。 */ @Transactional public ApiPullRuntimeResponse reset(Integer taskId, ResetApiCheckpointRequest request) { diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java index 45120fd..347e463 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java @@ -5,6 +5,7 @@ import com.fasterxml.jackson.core.JsonToken; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.fasterxml.jackson.databind.node.TextNode; import com.haobang.apipull.ApiPullProperties; import com.haobang.apipull.adapter.AdapterException; import com.haobang.apipull.adapter.ApiPullAdapter; @@ -193,13 +194,12 @@ public class WeibuApiAdapter implements ApiPullAdapter { ObjectNode body = json.createObjectNode(); body.putArray("assets_group"); - body.put("search_connection", false); - body.putArray("net_data_type").add("attack").add("risk"); body.put("time_from", vendorStart.getEpochSecond()); body.put("time_to", vendorEnd.getEpochSecond()); + // 不设置 net_data_type 或 threat.level 条件：该接口中的 attack、action、risk + // 都属于需要采集的逐条日志，具体取舍由下游设备泛化规则完成。 body.put("sql", ""); body.putArray("columns"); - body.put("asc", false); // 厂商接口只有 size，没有 offset/page。Runner 通过 total 自动缩小窗口，确保该次 // size 足以取回窗口内全部日志，而不是把 size 当作伪分页游标。 body.put("size", request.getLimit()); @@ -263,6 +263,149 @@ public class WeibuApiAdapter implements ApiPullAdapter { } } + /** + * 使用任务保存的地址和凭据执行一次只读诊断查询。 + * + *

请求 JSON 原样发送给微步，便于现场试验 {@code size}、{@code page}、{@code pageSize}、 + * {@code cur_page} 等不同版本可能支持的字段。调用者不能覆盖目标 URL 或认证参数；本方法也不写 + * WAL、不推进 checkpoint。请求体限制为 64 KiB，响应继续服从任务的 maxResponseBytes 和超时配置。 + */ + public DiagnosticResult diagnosticQuery(DeviceCollectApiConfig config, JsonNode requestBody) { + validateConfig(config); + if (requestBody == null || !requestBody.isObject()) fail("DIAGNOSTIC_REQUEST_INVALID"); + + byte[] requestBytes = null; + byte[] responseBytes = null; + HttpPost post = null; + long started = System.nanoTime(); + try { + requestBytes = json.writeValueAsBytes(requestBody); + if (requestBytes.length > 65536) fail("DIAGNOSTIC_REQUEST_TOO_LARGE"); + try (CredentialSnapshot snapshot = + credentials.load(config.getCredentialRef(), TYPE, config.getDeviceId())) { + char[] keyChars = snapshot.require("app_key").copy(); + char[] secretChars = snapshot.require("app_secret").copy(); + try { + String apiKey = new String(keyChars); + CredentialSchemas.validate(TYPE, apiKey, new String(secretChars), null); + long timestamp = Instant.now().getEpochSecond(); + URI endpoint = + new URIBuilder(endpoint(urls.validate(config.getBaseUrl()))) + .addParameter("api_key", apiKey) + .addParameter("auth_timestamp", Long.toString(timestamp)) + .addParameter("sign", sign(apiKey, secretChars, timestamp)) + .build(); + post = new HttpPost(endpoint); + post.setConfig(requestConfig(config)); + post.setHeader("Accept", "application/json"); + post.setEntity( + new StringEntity( + new String(requestBytes, StandardCharsets.UTF_8), + ContentType.APPLICATION_JSON)); + try (CloseableHttpResponse response = http.execute(post)) { + int status = response.getStatusLine().getStatusCode(); + responseBytes = readBounded(response.getEntity(), config.getMaxResponseBytes()); + JsonNode responseBody = parseDiagnosticBody(responseBytes); + return new DiagnosticResult( + status, + elapsed(started), + responseBytes.length, + redactSensitiveFields(responseBody)); + } + } finally { + Arrays.fill(keyChars, '\0'); + Arrays.fill(secretChars, '\0'); + } + } + } catch (AdapterException | CredentialException e) { + throw e; + } catch (SocketTimeoutException | org.apache.http.conn.ConnectTimeoutException e) { + fail("HTTP_TIMEOUT"); + return null; + } catch (IOException e) { + fail("DEVICE_API_UNAVAILABLE"); + return null; + } catch (Exception e) { + fail("REQUEST_BUILD_ERROR"); + return null; + } finally { + if (post != null) post.releaseConnection(); + if (requestBytes != null) Arrays.fill(requestBytes, (byte) 0); + if (responseBytes != null) Arrays.fill(responseBytes, (byte) 0); + } + } + + private JsonNode parseDiagnosticBody(byte[] bytes) { + try { + return parseBounded(bytes); + } catch (AdapterException invalidJson) { + // HTTP 错误页可能不是 JSON；诊断接口仍返回受大小限制的原始文本，便于现场定位网关问题。 + return TextNode.valueOf(new String(bytes, StandardCharsets.UTF_8)); + } + } + + private JsonNode redactSensitiveFields(JsonNode value) { + JsonNode copy = value.deepCopy(); + redactSensitiveFieldsInPlace(copy); + return copy; + } + + private void redactSensitiveFieldsInPlace(JsonNode value) { + if (value == null) return; + if (value.isObject()) { + ObjectNode object = (ObjectNode) value; + java.util.Iterator names = object.fieldNames(); + java.util.List fields = new java.util.ArrayList<>(); + while (names.hasNext()) fields.add(names.next()); + for (String field : fields) { + if (isSensitiveField(field)) object.put(field, "[REDACTED]"); + else redactSensitiveFieldsInPlace(object.get(field)); + } + } else if (value.isArray()) { + for (JsonNode item : value) redactSensitiveFieldsInPlace(item); + } + } + + private boolean isSensitiveField(String field) { + String normalized = field == null ? "" : field.toLowerCase(java.util.Locale.ROOT); + return "api_key".equals(normalized) + || "app_key".equals(normalized) + || "app_secret".equals(normalized) + || "sign".equals(normalized) + || "authorization".equals(normalized) + || "token".equals(normalized); + } + + public static final class DiagnosticResult { + private final int httpStatus; + private final long durationMs; + private final int responseBytes; + private final JsonNode response; + + public DiagnosticResult(int httpStatus, long durationMs, int responseBytes, JsonNode response) { + this.httpStatus = httpStatus; + this.durationMs = durationMs; + this.responseBytes = responseBytes; + this.response = response; + } + + public int getHttpStatus() { + return httpStatus; + } + + public long getDurationMs() { + return durationMs; + } + + public int getResponseBytes() { + return responseBytes; + } + + public JsonNode getResponse() { + return response; + } + } + private PullPage parsePage( byte[] bytes, int status, diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/ApiPullDtos.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/ApiPullDtos.java index 42c326d..3751265 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/ApiPullDtos.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/ApiPullDtos.java @@ -1,6 +1,7 @@ package com.haobang.apipull.api; import com.fasterxml.jackson.annotation.JsonAnySetter; +import com.fasterxml.jackson.databind.JsonNode; import java.time.OffsetDateTime; import java.util.Collections; import java.util.List; @@ -314,6 +315,18 @@ public final class ApiPullDtos { private ApiPullRunResponse latestRun; } + /** 微步只读诊断结果；不包含签名 URL、API Key、Secret 或其他认证信息。 */ + @Data + public static class WeibuDiagnosticQueryResponse { + private Integer taskId; + private Integer deviceId; + private long configVersion; + private int httpStatus; + private long durationMs; + private int responseBytes; + private JsonNode response; + } + @Data public static class ApiPullRunResponse { private String runId; diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/DeviceApiPullTaskController.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/DeviceApiPullTaskController.java index 42843b0..9086fb8 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/DeviceApiPullTaskController.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/api/DeviceApiPullTaskController.java @@ -3,6 +3,7 @@ package com.haobang.apipull.api; import com.haobang.apipull.ApiPullTaskService; import com.haobang.apipull.api.ApiPullDtos.*; import com.haobang.apipull.persistence.DeviceCollectApiRun; +import com.fasterxml.jackson.databind.JsonNode; import javax.validation.Valid; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; @@ -83,6 +84,13 @@ public class DeviceApiPullTaskController { return service.runs(taskId, page, pageSize); } + /** 请求体按原样发送给微步，便于试验现场版本支持的查询和分页字段。 */ + @PostMapping("/{taskId}/diagnostics/weibu-query") + public WeibuDiagnosticQueryResponse diagnosticWeibuQuery( + @PathVariable Integer taskId, @RequestBody JsonNode requestBody) { + return service.diagnosticWeibuQuery(taskId, requestBody); + } + @PostMapping("/{taskId}/checkpoint/reset") public ApiPullRuntimeResponse reset( @PathVariable Integer taskId, @Valid @RequestBody ResetApiCheckpointRequest r) { diff --git a/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceCollectApiMapper.xml b/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceCollectApiMapper.xml index 09de44c..baf597c 100644 --- a/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceCollectApiMapper.xml +++ b/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceCollectApiMapper.xml @@ -27,10 +27,15 @@ SELECT * FROM device_collect_api_checkpoint WHERE collect_task_id=#{taskId} UPDATE device_collect_api_config SET base_url=#{baseUrl},credential_ref=#{credentialRef}, @@ -56,7 +61,11 @@ fencing_token=fencing_token+1,updated_at=SYSDATE WHERE collect_task_id=#{taskId} AND (lease_owner IS NULL OR lease_expires_at < SYSDATE) AND EXISTS (SELECT 1 FROM device_collect_api_config c + INNER JOIN device_device d ON d.id=c.device_id WHERE c.collect_task_id=device_collect_api_checkpoint.collect_task_id AND c.enabled=1 + AND d.device_collect_id=#{deviceCollectId} + AND d.deleted_at IS NULL + AND (d.del_flag IS NULL OR d.del_flag='0') AND c.config_version=#{configVersion} AND (c.next_run_at IS NULL OR c.next_run_at <= SYSDATE)) diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullContractTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullContractTest.java index 480847a..ca70490 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullContractTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/apipull/ApiPullContractTest.java @@ -7,6 +7,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.haobang.apipull.api.ApiPullDtos.ApiPullTaskResponse; import com.haobang.apipull.api.ApiPullDtos.CreateApiPullTaskRequest; import java.io.InputStream; +import java.util.Collections; import org.apache.ibatis.builder.xml.XMLMapperBuilder; import org.apache.ibatis.session.Configuration; import org.junit.jupiter.api.Test; @@ -67,5 +68,25 @@ class ApiPullContractTest { assertTrue( configuration.hasStatement("com.common.mapper.DeviceApiCredentialMapper.selectForLoad")); assertTrue(configuration.hasStatement("com.common.mapper.DeviceApiCredentialMapper.updateCas")); + + String dueSql = + configuration + .getMappedStatement("com.common.mapper.DeviceCollectApiMapper.selectDueConfigs") + .getBoundSql(Collections.singletonMap("deviceCollectId", 1)) + .getSql() + .replaceAll("\\s+", " ") + .toLowerCase(); + assertTrue(dueSql.contains("inner join device_device d on d.id=c.device_id")); + assertTrue(dueSql.contains("d.device_collect_id=?")); + + String leaseSql = + configuration + .getMappedStatement("com.common.mapper.DeviceCollectApiMapper.acquireLease") + .getBoundSql(Collections.singletonMap("deviceCollectId", 1)) + .getSql() + .replaceAll("\\s+", " ") + .toLowerCase(); + assertTrue(leaseSql.contains("inner join device_device d on d.id=c.device_id")); + assertTrue(leaseSql.contains("d.device_collect_id=?")); } } 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 2d672c4..f70894d 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 @@ -21,14 +21,15 @@ class ApiPullSchedulerTest { DeviceCollectApiMapper mapper=mock(DeviceCollectApiMapper.class);ApiPullRunner runner=mock(ApiPullRunner.class);SyslogWalService wal=mock(SyslogWalService.class); when(wal.isDiskCritical()).thenReturn(false);when(wal.shouldPauseTcp()).thenReturn(false); DeviceCollectApiConfig first=config(1,"https://localhost/ndr/api");DeviceCollectApiConfig second=config(2,"https://localhost/ndr/api"); - when(mapper.selectConfig(1)).thenReturn(first);when(mapper.selectConfig(2)).thenReturn(second);when(mapper.acquireLease(eq(1),anyLong(),anyString(),anyInt())).thenReturn(1); + when(mapper.selectConfig(1)).thenReturn(first);when(mapper.selectConfig(2)).thenReturn(second);when(mapper.acquireLease(eq(1),anyLong(),anyString(),anyInt(),eq(1))).thenReturn(1); 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,resolver(DeviceCollectionMode.BOTH));scheduler.init(); + AppProperties app=appProperties(1); + ApiPullScheduler scheduler=new ApiPullScheduler(p,mapper,runner,wal,resolver(DeviceCollectionMode.BOTH,1,app),app);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());} + verify(mapper,never()).acquireLease(eq(2),anyLong(),anyString(),anyInt(),anyInt());} finally{release.countDown();scheduler.shutdown();} } @@ -37,21 +38,38 @@ class ApiPullSchedulerTest { 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(); + when(mapper.selectDueConfigs(1)).thenReturn(java.util.Collections.singletonList(config)); + AppProperties app=appProperties(1); + ApiPullScheduler scheduler=new ApiPullScheduler(p,mapper,runner,wal,resolver(DeviceCollectionMode.SYSLOG_ONLY,1,app),app);scheduler.init(); try{scheduler.scan(); verify(mapper).scheduleNext(1,1L,10); - verify(mapper,never()).acquireLease(anyInt(),anyLong(),anyString(),anyInt()); + verify(mapper,never()).acquireLease(anyInt(),anyLong(),anyString(),anyInt(),anyInt()); verifyNoInteractions(runner); }finally{scheduler.shutdown();} } - private DeviceContextResolver resolver(DeviceCollectionMode mode){ + @Test void reassignedTaskIsIgnoredWithoutAdvancingNextRunOrAcquiringLease(){ + 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(1)).thenReturn(java.util.Collections.singletonList(config)); + AppProperties app=appProperties(1); + ApiPullScheduler scheduler=new ApiPullScheduler(p,mapper,runner,wal,resolver(DeviceCollectionMode.BOTH,2,app),app);scheduler.init(); + try{scheduler.scan(); + verify(mapper).selectDueConfigs(1); + verify(mapper,never()).scheduleNext(anyInt(),anyLong(),anyInt()); + verify(mapper,never()).acquireLease(anyInt(),anyLong(),anyString(),anyInt(),anyInt()); + verifyNoInteractions(runner); + }finally{scheduler.shutdown();} + } + + private DeviceContextResolver resolver(DeviceCollectionMode mode,int deviceCollectId,AppProperties app){ 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); + when(devices.getByIdSafely(anyInt())).thenAnswer(i->{DeviceDevice d=new DeviceDevice();d.setId(i.getArgument(0));d.setDeviceCollectId(deviceCollectId);d.setCollectionMode(mode);return d;}); return new DeviceContextResolver(devices,mock(DeviceUnknownService.class),app); } + private AppProperties appProperties(int deviceCollectId){AppProperties app=new AppProperties();app.setAppServiceDeviceCollectId(deviceCollectId);app.setSyslogDeviceContextCacheTtlMs(60000L);return 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 a1632d6..db95334 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 @@ -17,14 +17,17 @@ import com.common.mapper.DeviceCollectApiMapper; import com.common.mapper.DeviceCollectTaskMapper; import com.common.service.DeviceDeviceService; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.haobang.apipull.adapter.AdapterException; import com.haobang.apipull.adapter.AdapterRegistry; import com.haobang.apipull.adapter.ApiPullAdapter; +import com.haobang.apipull.adapter.weibu.WeibuApiAdapter; import com.haobang.apipull.api.ApiPullDtos.ApiPullTaskResponse; import com.haobang.apipull.api.ApiPullDtos.CreateApiPullTaskRequest; import com.haobang.apipull.api.ApiPullDtos.UpdateApiPullTaskRequest; import com.haobang.apipull.api.ApiPullDtos.ValidateApiPullTaskResponse; import com.haobang.apipull.api.ApiPullDtos.ValidationCheckResponse; +import com.haobang.apipull.api.ApiPullDtos.WeibuDiagnosticQueryResponse; import com.haobang.apipull.credential.CredentialProvider; import com.haobang.apipull.credential.CredentialSnapshot; import com.haobang.apipull.persistence.DeviceCollectApiCheckpoint; @@ -334,6 +337,62 @@ class ApiPullTaskServiceTest { assertEquals(100, updated.getValue().getSchedulerPriority()); } + @Test + void weibuDiagnosticUsesTaskBindingWithoutChangingTaskState() { + DeviceCollectApiMapper apiMapper = mock(DeviceCollectApiMapper.class); + DeviceCollectApiConfig config = new DeviceCollectApiConfig(); + config.setCollectTaskId(21); + config.setDeviceId(40); + config.setAdapterType(WeibuApiAdapter.TYPE); + config.setConfigVersion(2L); + when(apiMapper.selectConfig(21)).thenReturn(config); + + DeviceDeviceService devices = mock(DeviceDeviceService.class); + DeviceDevice device = new DeviceDevice(); + device.setId(40); + device.setDeviceCollectId(7); + device.setCollectionMode(DeviceCollectionMode.API_ONLY); + when(devices.getByIdSafely(40)).thenReturn(device); + + ObjectMapper objectMapper = new ObjectMapper(); + ObjectNode requestBody = objectMapper.createObjectNode().put("page", 2).put("pageSize", 100); + ObjectNode vendorResponse = objectMapper.createObjectNode().put("response_code", 0); + WeibuApiAdapter adapter = mock(WeibuApiAdapter.class); + when(adapter.diagnosticQuery(config, requestBody)) + .thenReturn(new WeibuApiAdapter.DiagnosticResult(200, 18L, 123, vendorResponse)); + AdapterRegistry adapters = mock(AdapterRegistry.class); + when(adapters.require(WeibuApiAdapter.TYPE)).thenReturn(adapter); + + AppProperties app = new AppProperties(); + app.setAppServiceDeviceCollectId(7); + ApiPullProperties properties = new ApiPullProperties(); + ApiPullTaskService service = + new ApiPullTaskService( + mock(DeviceCollectTaskMapper.class), + apiMapper, + devices, + adapters, + mock(CredentialProvider.class), + new SafeBaseUrlValidator(properties), + mock(ApiPullScheduler.class), + app, + properties, + objectMapper); + + WeibuDiagnosticQueryResponse response = service.diagnosticWeibuQuery(21, requestBody); + + assertEquals(21, response.getTaskId()); + assertEquals(40, response.getDeviceId()); + assertEquals(2L, response.getConfigVersion()); + assertEquals(200, response.getHttpStatus()); + assertEquals(18L, response.getDurationMs()); + assertEquals(123, response.getResponseBytes()); + assertEquals(0, response.getResponse().path("response_code").asInt()); + verify(adapter).diagnosticQuery(config, requestBody); + verify(apiMapper, never()).insertRun(any()); + verify(apiMapper, never()).updateRun(any()); + } + private CreateApiPullTaskRequest request() { CreateApiPullTaskRequest request = new CreateApiPullTaskRequest(); request.setDeviceId(126); 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 index d31c57f..a24049e 100644 --- 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 @@ -90,12 +90,11 @@ class WeibuApiAdapterTest { 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("search_connection")); + assertFalse(body.has("net_data_type")); + assertFalse(body.has("asc")); assertFalse(body.has("condition")); assertFalse(body.has("page")); @@ -107,6 +106,46 @@ class WeibuApiAdapterTest { assertEquals(f.items().get(0).toString(), page.getEvents().get(0).getRawMessage()); } + @Test + void diagnosticQueryForwardsCustomPaginationFieldsAndRedactsSecrets() throws Exception { + Fixture f = new Fixture(); + ObjectNode requestBody = JSON.createObjectNode(); + requestBody.put("time_from", START.getEpochSecond()); + requestBody.put("time_to", END.getEpochSecond() - 1L); + requestBody.put("sql", ""); + requestBody.put("page", 3); + requestBody.put("pageSize", 17); + requestBody.put("size", 17); + f.root.put("api_key", "must-not-be-returned"); + + WeibuApiAdapter.DiagnosticResult result = + f.adapter.diagnosticQuery(f.config, requestBody); + + assertEquals(requestBody.toString(), f.body.get().toString()); + assertEquals(200, result.getHttpStatus()); + assertTrue(result.getDurationMs() >= 0); + assertTrue(result.getResponseBytes() > 0); + assertEquals("[REDACTED]", result.getResponse().path("api_key").asText()); + assertEquals(2, result.getResponse().path("data").path("total").asInt()); + Map signedQuery = query(f.request.get()); + assertEquals(API_KEY, signedQuery.get("api_key")); + assertNotNull(signedQuery.get("auth_timestamp")); + assertNotNull(signedQuery.get("sign")); + } + + @Test + void diagnosticQueryRejectsNonObjectBodyBeforeHttp() throws Exception { + Fixture f = new Fixture(); + + AdapterException failure = + assertThrows( + AdapterException.class, + () -> f.adapter.diagnosticQuery(f.config, JSON.createArrayNode())); + + assertEquals("DIAGNOSTIC_REQUEST_INVALID", failure.getCode()); + verifyNoInteractions(f.http); + } + @Test void stableLogIdIsIdempotentAndOffsetPaginationIsRejected() throws Exception { Fixture f = new Fixture(); @@ -284,8 +323,8 @@ class WeibuApiAdapterTest { .put("time", START.plusSeconds(200).getEpochSecond()) .put("direction", "lateral") .putObject("threat") - .put("level", "risk") - .put("name", "synthetic risk two"); + .put("level", "action") + .put("name", "synthetic action two"); return root; } diff --git a/haobang-security-dm/upgrade-doc/v6/README.md b/haobang-security-dm/upgrade-doc/v6/README.md index 16cb8f1..dc9d8d6 100644 --- a/haobang-security-dm/upgrade-doc/v6/README.md +++ b/haobang-security-dm/upgrade-doc/v6/README.md @@ -32,6 +32,19 @@ database/03-device-collection-mode-verify.sql 默认 `BOTH` 用于兼容现有设备；升级 SQL 不硬编码任何设备 ID。 +## API 拉取任务的探针归属 + +新版调度器只扫描满足以下条件的设备任务： + +```text +DEVICE_DEVICE.DEVICE_COLLECT_ID = 当前节点 APP_SERVICE_DEVICE_COLLECT_ID +``` + +取得数据库租约时会再次检查该归属关系。设备迁移到其他探针后，原节点不会执行 HTTP 请求， +也不会推进该任务的 `NEXT_RUN_AT`。同一探针的多个同版本副本可以同时开启 +`DEVICE_API_ENABLED=true`，再由数据库租约保证同一任务只有一个实例执行。混合版本升级期间， +旧节点仍建议保持 `DEVICE_API_ENABLED=false`。 + 本次 v6 强制数据库结构变更只有 `DEVICE_DEVICE.COLLECTION_MODE`。微步 API 复用现有 `DEVICE_COLLECT_API_*` 基础表，不在本次升级中新增或修改这些表。新版微步适配器只支持 `WEIBU_LOG`（`/api/v1/log/searchBySql`）；开发阶段的 `WEIBU_INCIDENT` 从未作为现场版本发布，因此不需要现场数据迁移脚本。 diff --git a/haobang-security-dm/upgrade-doc/v6/api/README.md b/haobang-security-dm/upgrade-doc/v6/api/README.md index f6d8d90..dc915dd 100644 --- a/haobang-security-dm/upgrade-doc/v6/api/README.md +++ b/haobang-security-dm/upgrade-doc/v6/api/README.md @@ -25,7 +25,9 @@ API 相关升级只看本页。示例设备：`deviceId=40`、`IP=10.56.171.164` docker logs syslog-serve-dm 2>&1 | grep '\[SYSLOG_BUILD_READY\]' | tail -n 1 ``` - 必须看到 `apiPullEnabled=true`。内网默认 `apiTlsVerifyEnabled=false`。 + 必须看到 `codeRevision=WEIBU_DIAGNOSTIC_V20260917_66`、`apiPullEnabled=true`，并确认 + `deviceCollectId` 与设备 40 的 `DEVICE_DEVICE.DEVICE_COLLECT_ID` 一致。内网默认 + `apiTlsVerifyEnabled=false`。 ## 2. 配置微步 API 任务 @@ -100,7 +102,37 @@ docker exec syslog-serve-dm curl -fsS \ GET 必须返回 `collectionMode=API_ONLY`。此后 `10.56.171.164` 的 TCP/UDP Syslog 被丢弃，API 任务继续运行。 -## 4. 验收 +## 4. 微步原始查询诊断 + +诊断接口复用任务中的设备地址、凭据、TLS 和超时配置，只把请求 JSON 原样发送给微步。 +它不会写入 WAL/Kafka，不会推进 checkpoint，也不会在响应中返回 API Key、Secret 或签名 URL。 + +查询最近 10 分钟，单次最多请求 1000 条： + +```bash +TIME_TO=$(date +%s) +TIME_FROM=$((TIME_TO - 600)) + +docker exec syslog-serve-dm curl -fsS -X POST \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks/21/diagnostics/weibu-query' \ + -H 'Content-Type: application/json' \ + -d "{\"time_from\":${TIME_FROM},\"time_to\":${TIME_TO},\"sql\":\"\",\"assets_group\":[],\"columns\":[],\"size\":1000}" +``` + +请求体不会限制微步字段，因此可以试验现场版本的页码和页大小字段。例如： + +```bash +docker exec syslog-serve-dm curl -fsS -X POST \ + 'http://127.0.0.1:8189/syslogserve/api/api-pull-tasks/21/diagnostics/weibu-query' \ + -H 'Content-Type: application/json' \ + -d "{\"time_from\":${TIME_FROM},\"time_to\":${TIME_TO},\"sql\":\"\",\"assets_group\":[],\"columns\":[],\"page\":2,\"pageSize\":100,\"size\":100}" +``` + +微步当前文档只明确描述了查询条件，现场已有脚本确认 `size` 可用，但未确认真正的页码字段。 +诊断接口会原样传递 `page`、`pageSize`、`cur_page`、`page_size` 等字段；应通过不同页返回的 +`response.data.data` 和 `response.data.total` 判断设备是否接受，不能仅凭 HTTP 200 认定分页生效。 + +## 5. 验收 ```bash docker exec syslog-serve-dm curl -fsS \ @@ -116,7 +148,7 @@ docker exec syslog-serve-dm curl -fsS \ - Kafka 中每条消息是一条微步 JSON，下游按设备 40 关联的泛化规则标准化。 - 来自 `10.56.171.164` 的 Syslog 不进入 WAL/Kafka，其他默认 `BOTH` 设备不受影响。 -## 5. 缓存清理 +## 6. 缓存清理 ```bash # syslog-serve：指定设备 / 全部设备 @@ -134,7 +166,7 @@ docker exec syslog-consumer-rule-dm curl -fsS -X DELETE \ 规则缓存清理会通知所有正常订阅 Redis 消息的 `syslog-consumer-rule` 实例。不要使用旧的 `DELETE /api/cache/all`。 -## 6. 快速回退 +## 7. 快速回退 1. 暂停微步 API 任务。 2. 将设备改回 `BOTH`：