diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java index 0e6c579..d51e83d 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java @@ -944,6 +944,11 @@ public class LogNormalProcessor { (HashMap) ((HashMap) map.get("action")).get("param"); normalColumMap.put("dest_field_value", getMappingValue(action_param, entry.getValue().toString())); + } else if (((HashMap) map.get("action")).get("type").equals("bounded_text")) { + Map params = (Map) + ((Map) map.get("action")).get("param"); + int bytes = com.common.util.BoundedText.limit(params.get("max_utf8_bytes")); + normalColumMap.put("dest_field_value", com.common.util.BoundedText.map(entry.getValue(), bytes)); } else if (((HashMap) map.get("action")).get("type").equals("time")) { normalColumMap.put("action_param", ((HashMap) map.get("action")).get("param")); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java index db78ef4..e8b3334 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java @@ -8,7 +8,7 @@ import java.util.Map; /** 单个字段映射的不可变执行计划。 */ public final class CompiledMapper { - public enum Action { EQUAL, MAPPING, TIME, LEGACY_NULL } + public enum Action { EQUAL, MAPPING, TIME, BOUNDED_TEXT, LEGACY_NULL } private final String originField; private final String destinationField; @@ -33,6 +33,9 @@ public final class CompiledMapper { case MAPPING: // 保持旧执行器语义：源值为 null 时规则执行失败，而不是静默映射成字符串 "null"。 return sanitize(mapping.get(sourceValue.toString())); + case BOUNDED_TEXT: + return sanitize(com.common.util.BoundedText.map(sourceValue, + com.common.util.BoundedText.limit(mapping.get("max_utf8_bytes")))); case TIME: try { return TimeConversionUtils.convertToMillis(sourceValue.toString(), timezone); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java index 323af22..2acb088 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java @@ -144,6 +144,13 @@ public class CompiledRuleCompiler { mapping.put(entry.getKey(), freeze(objectMapper.convertValue(entry.getValue(), Object.class))); } compiled = new CompiledMapper(origin, destination, CompiledMapper.Action.MAPPING, mapping, null); + } else if ("bounded_text".equals(type)) { + JsonNode limit = param == null ? null : param.get("max_utf8_bytes"); + if (limit == null || !limit.isIntegralNumber() || !limit.canConvertToInt()) + throw invalid(source, "bounded_text requires integer max_utf8_bytes"); + int bytes = com.common.util.BoundedText.limit(limit.intValue()); + compiled = new CompiledMapper(origin, destination, CompiledMapper.Action.BOUNDED_TEXT, + Collections.singletonMap("max_utf8_bytes", bytes), null); } else if ("time".equals(type)) { String timezone = param == null ? null : text(param.get("timezone")); compiled = new CompiledMapper(origin, destination, CompiledMapper.Action.TIME, diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/BoundedText.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/BoundedText.java new file mode 100644 index 0000000..91cc097 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/BoundedText.java @@ -0,0 +1,32 @@ +package com.common.util; + +/** Explicit opt-in rule action: keep text only within its UTF-8 budget, never truncate or decode. */ +public final class BoundedText { + private BoundedText() { } + + public static int limit(Object value) { + if (!(value instanceof Number)) throw new IllegalArgumentException("max_utf8_bytes must be an integer"); + Number n = (Number) value; + int limit = n.intValue(); + if (n.doubleValue() != limit || limit < 1 || limit > 1048576) + throw new IllegalArgumentException("max_utf8_bytes out of range"); + return limit; + } + + public static String map(Object value, int maxBytes) { + if (!(value instanceof String)) return null; + String text = (String) value; + int bytes = 0; + for (int i = 0; i < text.length(); i++) { + char c = text.charAt(i); + if (c <= 0x7f) bytes++; + else if (c <= 0x7ff) bytes += 2; + else if (Character.isHighSurrogate(c) && i + 1 < text.length() + && Character.isLowSurrogate(text.charAt(i + 1))) { bytes += 4; i++; } + else if (Character.isSurrogate(c)) return null; + else bytes += 3; + if (bytes > maxBytes) return null; + } + return text; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/QuanxiNormalizationTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/QuanxiNormalizationTest.java new file mode 100644 index 0000000..288f322 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/QuanxiNormalizationTest.java @@ -0,0 +1,165 @@ +package com.Modules.NormalData; + +import com.Modules.NormalData.compiled.*; +import com.common.mapper.*; +import com.common.service.*; +import com.config.*; +import com.fasterxml.jackson.databind.*; +import com.fasterxml.jackson.databind.node.*; +import com.common.util.BoundedText; +import java.nio.file.*; +import java.time.*; +import java.time.format.DateTimeFormatter; +import java.util.*; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.mockito.ArgumentMatchers.anyMap; + +/** Shared sanitized vendor response + adapter golden identities -> real normalization -> real routing. */ +class QuanxiNormalizationTest { + static final ObjectMapper JSON = new ObjectMapper(); + static final LocalDateTime RECEIVED = LocalDateTime.of(2026, 9, 6, 12, 0); + static final DateTimeFormatter TIME = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS"); + + @Test void allTwentyDocumentsNormalizeAndUseAlarmInsertInBothEngines() throws Exception { + JsonNode records = JSON.readTree(file("fixtures/api-response.json").toFile()).path("result").path("data"); + JsonNode expected = JSON.readTree(file("fixtures/expected-events.json").toFile()); + for (boolean compiled : new boolean[]{false, true}) { + Fixture f = new Fixture(compiled); + for (int i = 0; i < records.size(); i++) { + JsonNode r = records.get(i); + Map row = f.prepare(r, expected.get(i).path("event_id").asText()); + assertEquals(r.path("src_ip").asText(), row.get("src_ip")); + assertEquals(r.path("dest_ip").asText(), row.get("dest_ip")); + assertEquals(r.path("src_port").asInt(), row.get("src_port")); + assertEquals(r.path("dest_port").asInt(), row.get("dest_port")); + assertEquals(r.path("proto").asText(), row.get("proto")); + assertEquals(r.path("app_proto").asText(), row.get("protocol")); + assertEquals(r.path("attacker").asText(), row.get("attacker_ip")); + assertEquals(r.path("victim").asText(), row.get("victim_ip")); + assertEquals(r.path("direction").asText(), row.get("direction")); + assertEquals(r.path("name").asText(), row.get("origin_event_name")); + assertEquals(r.path("type").asText(), row.get("origin_event_type")); + assertEquals(r.path("severity").asInt(), row.get("origin_event_level")); + assertEquals(r.path("result").asText(), row.get("origin_attack_result")); + assertEquals(r.path("doc_id").asText(), row.get("origin_event_id")); + assertEquals(r.path("flow_id").asText(), row.get("tc_flow_id")); + assertEquals(-1, row.get("event_level")); // Explicit unknown, not guessed vendor severity. + assertEquals(-1, row.get("attack_result")); + assertEquals(LocalDateTime.ofInstant(Instant.ofEpochMilli(r.path("timestamp").asLong()), + ZoneId.of("Asia/Shanghai")), row.get("log_time")); + assertEquals(RECEIVED, row.get("created_at")); + assertEquals(row.get("id"), f.prepare(r, expected.get(i).path("event_id").asText()).get("id")); + JsonNode h = r.path("appbrief").path("http"); + for (String[] field : HTTP_FIELDS) { + assertEquals(h.has(field[0]) ? h.get(field[0]).asText().trim() : null, row.get(field[1])); + } + assertTrue(f.store.isAlarmData(row)); + assertEquals(1, f.store.insertDynamic(row)); + } + verify(f.alarm, times(20)).insertDynamic(anyMap()); + verifyNoInteractions(f.normal); + } + } + + @Test void missingWrongTypesOversizeAndExistingTopLevelFieldsAreHandledWithoutChangingRaw() throws Exception { + for (boolean compiled : new boolean[]{false, true}) { + Fixture f = new Fixture(compiled); + ObjectNode original = (ObjectNode) JSON.readTree(file("fixtures/api-response.json").toFile()) + .path("result").path("data").get(6); + for (String[] field : HTTP_FIELDS) { + for (JsonNode value : Arrays.asList(NullNode.instance, IntNode.valueOf(42), + JSON.createArrayNode().add("bad"), JSON.createObjectNode().put("bad", true), + TextNode.valueOf(repeat("x", 10001)))) { + ObjectNode record = original.deepCopy(); + record.with("appbrief").with("http").set(field[0], value); + record.put(field[1], "unrelated-existing-value"); + String before = record.toString(); + assertNull(f.prepare(record, "synthetic-id").get(field[1])); + assertEquals(before, record.toString()); + } + ObjectNode record = original.deepCopy(); + record.with("appbrief").with("http").put(field[0], repeat("x", 10000)); + assertEquals(10000, ((String) f.prepare(record, "synthetic-id").get(field[1])).length()); + record.with("appbrief").with("http").remove(field[0]); + assertNull(f.prepare(record, "synthetic-id").get(field[1])); + } + original.put("payload", "R0VUIC8gSFRUUC8xLjE="); + original.remove("appbrief"); + for (String[] field : HTTP_FIELDS) assertNull(f.prepare(original, "synthetic-id").get(field[1])); + } + } + + @Test void utf8BoundaryUsesBytesAndRejectsMalformedSurrogates() { + assertNotNull(BoundedText.map(repeat("中", 3333), 10000)); + assertNull(BoundedText.map(repeat("中", 3334), 10000)); + assertNotNull(BoundedText.map(repeat("😀", 2500), 10000)); + assertNull(BoundedText.map(repeat("😀", 2501), 10000)); + assertNull(BoundedText.map("\uD800", 10000)); + assertNull(BoundedText.map(new Object(), 10000)); + assertEquals("", BoundedText.map("", 10000)); + assertThrows(IllegalArgumentException.class, () -> BoundedText.limit(0)); + assertThrows(IllegalArgumentException.class, () -> BoundedText.limit(1.5)); + } + + @Test void ruleDestinationsAreActualAlarmSqlColumnsAndHttpPathsAreExact() throws Exception { + JsonNode rule = JSON.readTree(file("normalization-rule.json").toFile()); + String sql = new String(Files.readAllBytes(file("../../../syslog-consumer-rule/src/main/resources/mapper/SyslogNormalAlarmMapper.xml")), "UTF-8"); + for (JsonNode m : rule.path("mappers")) assertTrue(sql.contains("dataMap." + m.path("dest_field").asText() + " ")); + for (String[] field : HTTP_FIELDS) { + boolean found = false; + for (JsonNode m : rule.path("mappers")) if (field[1].equals(m.path("dest_field").asText())) { + assertEquals("appbrief.http." + field[0], m.path("origin_field").asText()); found = true; + } + assertTrue(found); + } + } + + static final String[][] HTTP_FIELDS = {{"req_header_raw", "http_req_header"}, {"req_body", "http_req_body"}, + {"resp_header_raw", "http_resp_header"}, {"resp_body", "http_resp_body"}}; + static String repeat(String s, int n) { StringBuilder b = new StringBuilder(); for (int i=0;i prepare(JsonNode doc, String id) { + LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(doc.path("timestamp").asLong()), ZoneId.of("Asia/Shanghai")); + String message = "[receive_time=" + TIME.format(RECEIVED) + " event_time=" + TIME.format(time) + + " event_id=" + id + " device_id=7 device_name=fixture vendor=quanxi data_type=json device_collect_id=3]" + doc; + LogNormalProcessor.NormalizationResult result = processor.prepare(message, id, "raw", RECEIVED, + snapshot, new BatchEnrichmentContext(false, tasks, devices), new RuleExecutionMetrics()); + assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome()); + assertEquals(1, result.getRows().size()); + return result.getRows().get(0); + } + } +} diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiCredentialService.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiCredentialService.java index f3a6130..ed1441c 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiCredentialService.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiCredentialService.java @@ -21,7 +21,7 @@ import org.springframework.transaction.annotation.Transactional; */ @Service public class ApiCredentialService { - private static final String NDRSERVER_V1 = "NDRSERVER_V1"; + private final DeviceApiCredentialMapper mapper; private final DeviceDeviceService devices; @@ -44,9 +44,8 @@ public class ApiCredentialService { public ApiCredentialResponse create(CreateApiCredentialRequest request) { requireOwnedDevice(request.getDeviceId()); adapters.require(request.getAdapterType()); - if (!NDRSERVER_V1.equals(request.getAdapterType())) { - throw error("CREDENTIAL_SCHEMA", "当前凭据字段只支持 NDRSERVER_V1", HttpStatus.BAD_REQUEST); - } + com.haobang.apipull.credential.CredentialSchemas.validate( + request.getAdapterType(), request.getAppKey(), request.getAppSecret(), request.getApiToken()); DeviceApiCredential credential = new DeviceApiCredential(); credential.setCredentialRef(request.getCredentialRef()); @@ -55,6 +54,7 @@ public class ApiCredentialService { credential.setCredentialVersion(request.getCredentialVersion()); credential.setAppKey(request.getAppKey()); credential.setAppSecret(request.getAppSecret()); + credential.setApiToken(request.getApiToken()); credential.setConfigVersion(1L); if (mapper.insert(credential) != 1) { throw error("INTERNAL_ERROR", "创建设备凭据失败", HttpStatus.INTERNAL_SERVER_ERROR); @@ -74,7 +74,11 @@ public class ApiCredentialService { DeviceApiCredential current = require(credentialRef); requireOwnedDevice(current.getDeviceId()); + com.haobang.apipull.credential.CredentialSchemas.validate( + current.getAdapterType(), request.getAppKey(), request.getAppSecret(), request.getApiToken()); DeviceApiCredential update = new DeviceApiCredential(); + update.setAdapterType(current.getAdapterType()); + update.setApiToken(request.getApiToken()); update.setCredentialRef(credentialRef); update.setCredentialVersion(request.getCredentialVersion()); update.setAppKey(request.getAppKey()); diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullRunner.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullRunner.java index 3450f01..2248c75 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullRunner.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullRunner.java @@ -117,7 +117,7 @@ public class ApiPullRunner { try { // safetyDelay 之后的数据仍可能在设备端持续变化，只处理已经闭合的逻辑窗口。 Instant start = ApiPullTime.toInstant(checkpoint.getCommittedWindowEnd()); - Instant now = Instant.now(); + Instant now = Instant.now().truncatedTo(java.time.temporal.ChronoUnit.MILLIS); Instant closedAt = now.minusSeconds(config.getSafetyDelaySeconds()); Instant plannedEnd = min(start.plusSeconds(config.getWindowSizeSeconds()), closedAt); if (!plannedEnd.isAfter(start)) { @@ -220,6 +220,9 @@ public class ApiPullRunner { long offset = 0; Long firstTotal = null; boolean complete = false; + String datasetFingerprint = firstPage.getDatasetFingerprint(); + java.util.Set seenExternalIds = adapter.requiresDistinctExternalIds() + ? new java.util.HashSet<>() : null; // 同一任务严格串行分页，offset 只在整页进入 WAL 后递增。 for (int pageNo = 1; pageNo <= config.getMaxPagesPerRun(); pageNo++) { @@ -247,14 +250,27 @@ public class ApiPullRunner { // offset 分页依赖稳定数据集；total 漂移时继续翻页可能静默漏数或重复。 if (firstTotal == null) firstTotal = page.getTotal(); else if (firstTotal.longValue() != page.getTotal()) - throw new AdapterException("PAGINATION_DATASET_CHANGED", "固定窗口内 NDR total 发生变化"); - if (page.getEvents().isEmpty()) { + throw new AdapterException("PAGINATION_DATASET_CHANGED", "固定窗口内分页 total 发生变化"); + if (!java.util.Objects.equals(datasetFingerprint, page.getDatasetFingerprint())) + throw new AdapterException("PAGINATION_DATASET_CHANGED", "固定窗口内分页元数据发生变化"); + int sourceRecordCount = page.getSourceRecordCount(); + if (sourceRecordCount < 0 || sourceRecordCount > config.getPageSize()) + throw new AdapterException("PAGINATION_INCONSISTENT", "设备单页返回条数超出请求 limit"); + if (seenExternalIds != null) { + for (PulledEvent event : page.getEvents()) { + if (seenExternalIds.size() >= 100000) + throw new AdapterException("PAGINATION_ID_LIMIT", "分页去重保护上限 100000"); + if (!seenExternalIds.add(event.getExternalEventId())) + throw new AdapterException("PAGINATION_DUPLICATE_DOCUMENT", "固定窗口出现重复文档 ID"); + } + } + if (sourceRecordCount == 0) { if (offset >= firstTotal) { complete = true; logSlowPage(config, run, page, pageOffset, 0, pageStarted); break; } - throw new AdapterException("PAGINATION_INCONSISTENT", "NDR 在 total 之前返回空页"); + throw new AdapterException("PAGINATION_INCONSISTENT", "设备在 total 之前返回空页"); } run.setRecordsFetched(run.getRecordsFetched() + page.getEvents().size()); List> receipts = new ArrayList<>(); @@ -298,9 +314,11 @@ public class ApiPullRunner { run.setRecordsDurable(run.getRecordsDurable() + 1L); } long walMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - walStarted); - offset += page.getEvents().size(); + // 适配器可能为了把厂商闭区间转换成平台半开区间而过滤边界记录。offset 必须按设备 + // 原始返回数推进，不能按过滤后的 WAL 写入数推进，否则会重复请求同一页。 + offset += sourceRecordCount; if (offset > firstTotal) - throw new AdapterException("PAGINATION_INCONSISTENT", "NDR 返回条数超过首次 total"); + throw new AdapterException("PAGINATION_INCONSISTENT", "设备返回条数超过首次 total"); mapper.updateRun(run); logSlowPage(config, run, page, pageOffset, walMs, pageStarted); if (offset >= firstTotal) { @@ -449,10 +467,11 @@ public class ApiPullRunner { long total = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started); if (total < properties.getSlowPageThresholdMs()) return; // 一条结构化日志覆盖网络、解析和 durable 三段耗时，便于定位千万级吞吐瓶颈。 - double rate = p.getEvents().isEmpty() ? 0d : p.getEvents().size() * 1000d / Math.max(1, total); + double rate = + p.getEvents().isEmpty() ? 0d : p.getEvents().size() * 1000d / Math.max(1, total); log.warn( "设备 API 慢页: taskId={}, runId={}, adapter={}, stream={}, windowStart={}, windowEnd={}," - + " offset={}, limit={}, count={}, responseBytes={}, httpMs={}, parseMs={}," + + " offset={}, limit={}, sourceCount={}, acceptedCount={}, responseBytes={}, httpMs={}, parseMs={}," + " walDurableMs={}, totalMs={}, recordsPerSecond={}", c.getCollectTaskId(), r.getRunId(), @@ -462,6 +481,7 @@ public class ApiPullRunner { r.getLogicalWindowEnd(), offset, c.getPageSize(), + p.getSourceRecordCount(), p.getEvents().size(), p.getResponseBytes(), p.getHttpDurationMs(), 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 c8d5a35..f989ce3 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 @@ -296,16 +296,17 @@ public class ApiPullTaskService { if (valid && adapter != null) { long pullStarted = System.nanoTime(); try { - Instant end = Instant.now().minusSeconds(c.getSafetyDelaySeconds()); + Instant end = Instant.now().truncatedTo(java.time.temporal.ChronoUnit.MILLIS) + .minusSeconds(c.getSafetyDelaySeconds()); Instant start = end.minusSeconds(1); // 验证仅做 limit=1 的闭合窗口查询，不推进 checkpoint，也不写入 WAL/Kafka。 adapter.pull(new PullRequest(c, start, end, 1, 0)); checks.add(check("TLS_HANDSHAKE", "PASS", null, "TLS 和主机名校验通过", pullStarted)); - checks.add(check("AUTHENTICATION", "PASS", null, "Challenge/Verify 认证通过", pullStarted)); + checks.add(check("AUTHENTICATION", "PASS", null, "厂商认证及查询通过", pullStarted)); checks.add(check("API_PERMISSION", "PASS", null, "业务接口可访问", pullStarted)); checks.add( check( - "NDR_ALERT".equals(c.getStreamKey()) ? "MINIMAL_ALERT_QUERY" : "MINIMAL_FLOW_QUERY", + minimalQueryCheck(c), "PASS", null, "最小查询通过", @@ -314,11 +315,16 @@ public class ApiPullTaskService { valid = false; String failureCode = code(e); String queryCheck = - "NDR_ALERT".equals(c.getStreamKey()) ? "MINIMAL_ALERT_QUERY" : "MINIMAL_FLOW_QUERY"; - if (failureCode.startsWith("QUERY_")) { + minimalQueryCheck(c); + if ("QUANXI_V1".equals(c.getAdapterType())) { + // A JSON-RPC error may itself be an auth/permission error. HTTP 200 is not proof of auth. + checks.add(check(queryCheck, "FAIL", failureCode, safe(e), pullStarted)); + checks.add(skipped("AUTHENTICATION")); + checks.add(skipped("API_PERMISSION")); + } else if (failureCode.startsWith("QUERY_")) { // QUERY_ 表示已完成 TLS、Challenge/Verify 并收到业务接口的 2xx 响应，不能误报为认证失败。 checks.add(check("TLS_HANDSHAKE", "PASS", null, "TLS 和主机名校验通过", pullStarted)); - checks.add(check("AUTHENTICATION", "PASS", null, "Challenge/Verify 认证通过", pullStarted)); + checks.add(check("AUTHENTICATION", "PASS", null, "厂商认证及查询通过", pullStarted)); checks.add(check("API_PERMISSION", "PASS", null, "业务接口可访问", pullStarted)); checks.add(check(queryCheck, "FAIL", failureCode, safe(e), pullStarted)); } else { @@ -333,7 +339,7 @@ public class ApiPullTaskService { ensureCheck(checks, "API_PERMISSION"); ensureCheck( checks, - "NDR_ALERT".equals(c.getStreamKey()) ? "MINIMAL_ALERT_QUERY" : "MINIMAL_FLOW_QUERY"); + minimalQueryCheck(c)); ValidateApiPullTaskResponse r = new ValidateApiPullTaskResponse(); r.setTaskId(taskId); r.setConfigVersion(version); @@ -608,4 +614,8 @@ public class ApiPullTaskService { private long n(Long v) { return v == null ? 0 : v; } + private String minimalQueryCheck(DeviceCollectApiConfig config) { + return "NDR_FLOW_LOG".equals(config.getStreamKey()) ? "MINIMAL_FLOW_QUERY" : "MINIMAL_ALERT_QUERY"; + } + } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ApiPullAdapter.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ApiPullAdapter.java index f61f1de..2c18825 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ApiPullAdapter.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ApiPullAdapter.java @@ -26,6 +26,9 @@ public interface ApiPullAdapter { */ int minimumWindowSeconds(); + /** Some offset protocols require distinct document IDs across a complete window. */ + default boolean requiresDistinctExternalIds() { return false; } + /** 在保存任务和执行请求前校验本厂商的全部约束。 */ void validateConfig(DeviceCollectApiConfig config); diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/PullPage.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/PullPage.java index f6caf43..1ea29a7 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/PullPage.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/PullPage.java @@ -9,7 +9,9 @@ import java.util.List; *

{@code total} 是当前固定查询窗口的数据集总数；执行器要求后续各页 total 保持一致， 防止 offset 分页过程中数据集漂移而静默漏数。耗时和字节数用于慢页日志。 */ public final class PullPage { + private final String datasetFingerprint; private final List events; + private final int sourceRecordCount; private final long total; private final int httpStatus; private final long responseBytes; @@ -23,7 +25,59 @@ public final class PullPage { long responseBytes, long httpDurationMs, long parseDurationMs) { - this.events = Collections.unmodifiableList(events); + this( + events, + total, + httpStatus, + responseBytes, + httpDurationMs, + parseDurationMs, + null, + events.size()); + } + + /** Optional vendor count/snapshot metadata, compared across pages by Runner. */ + public PullPage( + List events, + long total, + int httpStatus, + long responseBytes, + long httpDurationMs, + long parseDurationMs, + String datasetFingerprint) { + this( + events, + total, + httpStatus, + responseBytes, + httpDurationMs, + parseDurationMs, + datasetFingerprint, + events.size()); + } + + /** + * 创建允许适配器过滤厂商边界记录的结果页。 + * + *

{@code sourceRecordCount} 是设备本页实际返回的元素数，{@code events} 是适配器按通用逻辑窗口 + * 过滤后允许进入 WAL 的元素。Runner 必须按前者推进 offset，按后者写入 WAL；否则边界页被全部过滤时 + * 会重复请求同一页，或者错误地把设备响应判断为空页。 + */ + public PullPage( + List events, + long total, + int httpStatus, + long responseBytes, + long httpDurationMs, + long parseDurationMs, + String datasetFingerprint, + int sourceRecordCount) { + if (sourceRecordCount < events.size()) { + throw new IllegalArgumentException("sourceRecordCount 不能小于过滤后的事件数"); + } + this.datasetFingerprint = datasetFingerprint; + this.events = Collections.unmodifiableList(new java.util.ArrayList<>(events)); + this.sourceRecordCount = sourceRecordCount; this.total = total; this.httpStatus = httpStatus; this.responseBytes = responseBytes; @@ -31,10 +85,17 @@ public final class PullPage { this.parseDurationMs = parseDurationMs; } + public String getDatasetFingerprint() { return datasetFingerprint; } + public List getEvents() { return events; } + /** 设备本页实际返回的元素数，用于正确推进厂商 offset。 */ + public int getSourceRecordCount() { + return sourceRecordCount; + } + public long getTotal() { return total; } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapter.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapter.java index 01bf638..6fb1bc0 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapter.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/ndr/NdrServerApiAdapter.java @@ -25,6 +25,7 @@ import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatterBuilder; import java.time.temporal.ChronoField; +import java.time.temporal.ChronoUnit; import java.time.temporal.TemporalAccessor; import java.util.ArrayList; import java.util.Arrays; @@ -175,9 +176,16 @@ public class NdrServerApiAdapter implements ApiPullAdapter { URI base = urlValidator.validate(config.getBaseUrl()); String path = ALERT.equals(config.getStreamKey()) ? "/threatalert/investigate" : "/netlog/investigate"; + Instant vendorQueryStart = floorToSecond(request.getQueryStart()); + Instant vendorQueryEnd = ceilToSecond(request.getQueryEnd()); ObjectNode body = mapper.createObjectNode(); - body.put("start_time", QUERY_TIME.format(request.getQueryStart())); - body.put("end_time", QUERY_TIME.format(request.getQueryEnd())); + /* + * 平台窗口统一采用 [queryStart, queryEnd)，NDR 请求字段却只有秒精度，且现场接口会返回 + * 两端边界。请求范围向外扩到完整秒，响应后再按原始半开窗口过滤，既不会遗漏带微秒的 + * 事件，也不会让相邻窗口共同把边界事件写入 WAL。 + */ + body.put("start_time", QUERY_TIME.format(vendorQueryStart)); + body.put("end_time", QUERY_TIME.format(vendorQueryEnd)); if (FLOW.equals(config.getStreamKey())) { body.putArray("src_ip"); body.putArray("dest_ip"); @@ -223,11 +231,17 @@ public class NdrServerApiAdapter implements ApiPullAdapter { array = mapper.createArrayNode(); if (array == null || !array.isArray()) throw new AdapterException("RESPONSE_SCHEMA_ERROR", "NDR 响应缺少 data." + arrayName); + int sourceRecordCount = array.size(); ArrayList events = new ArrayList<>(); for (JsonNode item : array) { if (!item.isObject()) throw new AdapterException("RESPONSE_SCHEMA_ERROR", "NDR 日志元素不是对象"); String external = externalId(item, config.getStreamKey()); Instant eventTime = parseEventTime(text(item, "timestamp")); + // NDR 查询使用向外扩展的秒级闭区间；只有平台半开窗口内的事件可以进入 WAL。 + if (eventTime.isBefore(request.getQueryStart()) + || !eventTime.isBefore(request.getQueryEnd())) { + continue; + } if (ALERT.equals(config.getStreamKey())) addAlertCompatibilityAliases((ObjectNode) item); else addFlowCompatibilityAliases((ObjectNode) item); // 两级 UUIDv5 隔离任务/流：同一设备事件重拉得到相同 ID，不同任务或流不会碰撞。 @@ -244,7 +258,9 @@ public class NdrServerApiAdapter implements ApiPullAdapter { result.status, result.body.length, httpMs, - elapsed(parseStarted)); + elapsed(parseStarted), + null, + sourceRecordCount); } catch (AdapterException e) { // 此代码块只处理已通过认证且 HTTP 2xx 的业务查询响应，增加 QUERY_ 前缀供验证端准确归类阶段。 throw new AdapterException("QUERY_" + e.getCode(), e.getMessage(), e.getHttpStatus(), e); @@ -255,6 +271,17 @@ public class NdrServerApiAdapter implements ApiPullAdapter { } } + /** 向下取整到整秒，覆盖 queryStart 所在秒内可能晚于起点的微秒事件。 */ + static Instant floorToSecond(Instant value) { + return value.truncatedTo(ChronoUnit.SECONDS); + } + + /** 向上取整到整秒，覆盖 queryEnd 所在秒后再由本适配器执行右端开区间过滤。 */ + static Instant ceilToSecond(Instant value) { + Instant floor = floorToSecond(value); + return floor.equals(value) ? floor : floor.plusSeconds(1L); + } + private TokenLease token(DeviceCollectApiConfig config, URI base, boolean force) { try (CredentialSnapshot snapshot = credentials.load(config.getCredentialRef(), TYPE, config.getDeviceId())) { diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapter.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapter.java new file mode 100644 index 0000000..1925bc0 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/quanxi/QuanxiApiAdapter.java @@ -0,0 +1,320 @@ +package com.haobang.apipull.adapter.quanxi; + +import com.fasterxml.jackson.core.JsonParser; +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.haobang.apipull.adapter.*; +import com.haobang.apipull.credential.*; +import com.haobang.apipull.persistence.DeviceCollectApiConfig; +import com.haobang.apipull.security.SafeBaseUrlValidator; +import java.io.*; +import java.net.SocketTimeoutException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.time.Instant; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.util.*; +import java.util.concurrent.TimeUnit; +import javax.annotation.PreDestroy; +import org.apache.http.HttpEntity; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +/** + * 全悉 AlarmService.GetAlarmDocument JSON-RPC protocol. Each returned document is a snapshot, + * not a promise of an individual underlying alarm. Query semantics must be explicitly configured. + * Original vendor fields are retained, with no payload decoding or HTTP body duplication. + */ +@Component +public class QuanxiApiAdapter implements ApiPullAdapter { + public static final String TYPE = "QUANXI_V1"; + public static final String STREAM = "QUANXI_ALARM_DOCUMENT"; + private static final UUID NAMESPACE = UUID.fromString("736b8059-92a3-5ce5-bb0c-7a7b10e1d95c"); + private static final DateTimeFormatter ID_DATE = + DateTimeFormatter.ofPattern("yyyyMMdd").withZone(ZoneId.of("Asia/Shanghai")); + private final CredentialProvider credentials; + private final SafeBaseUrlValidator urls; + private final ObjectMapper json; + private final CloseableHttpClient http; + + @Autowired + public QuanxiApiAdapter(CredentialProvider credentials, SafeBaseUrlValidator urls, ObjectMapper json) { + this(credentials, urls, json, + HttpClients.custom().disableAutomaticRetries().disableRedirectHandling().build()); + } + + QuanxiApiAdapter(CredentialProvider credentials, SafeBaseUrlValidator urls, + ObjectMapper json, CloseableHttpClient http) { + this.credentials = credentials; + this.urls = urls; + this.json = json.copy().deactivateDefaultTyping() + .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION); + this.http = http; + } + + @Override public String type() { return TYPE; } + @Override public int version() { return 1; } + @Override public Set streams() { return Collections.singleton(STREAM); } + @Override public int minimumWindowSeconds() { return 1; } + @Override public boolean requiresDistinctExternalIds() { return true; } + + @Override public void validateConfig(DeviceCollectApiConfig c) { + if (c == null || !TYPE.equals(c.getAdapterType()) || !STREAM.equals(c.getStreamKey())) + throw failure("CONFIG_SCHEMA_ERROR"); + if (!"/rpc".equals(urls.validate(c.getBaseUrl()).getPath())) + throw failure("CONFIG_SCHEMA_ERROR"); + range(c.getPageSize(), 1, 1000); + range(c.getMaxPagesPerRun(), 1, 1000); + if ((long) c.getPageSize() * c.getMaxPagesPerRun() > 100000L + || (c.getInitialStartTime() != null && c.getInitialStartTime().getNano() % 1000000 != 0)) + throw failure("CONFIG_SCHEMA_ERROR"); + range(c.getPollIntervalSeconds(), 1, 86400); + range(c.getWindowSizeSeconds(), 1, 3600); + range(c.getLookbackSeconds(), 0, 86400); + range(c.getSafetyDelaySeconds(), 0, 86400); + range(c.getConnectTimeoutMs(), 100, 30000); + range(c.getRequestTimeoutMs(), 1000, 120000); + if (c.getMaxResponseBytes() == null || c.getMaxResponseBytes() < 1024 + || c.getMaxResponseBytes() > 67108864L) throw failure("CONFIG_SCHEMA_ERROR"); + options(c); + } + + private ObjectNode options(DeviceCollectApiConfig c) { + try { + JsonNode n = json.readTree(c.getExtraConfigJson()); + exactKeys(n, "recordMode", "paginationTotalField", "endInclusive", "queryParams"); + if (!"DOCUMENT_SNAPSHOT".equals(n.path("recordMode").asText()) + || !("total".equals(n.path("paginationTotalField").asText()) + || "page_total".equals(n.path("paginationTotalField").asText())) + || !n.path("endInclusive").isBoolean()) throw failure("CONFIG_SCHEMA_ERROR"); + JsonNode p = n.get("queryParams"); + // Explicit null means OMIT. No guessed default for these three vendor parameters. + exactKeys(p, "ignore", "collapse_key", "agg_sort"); + if (!p.get("ignore").isNull() && (!p.get("ignore").isIntegralNumber() + || !p.get("ignore").canConvertToInt())) throw failure("CONFIG_SCHEMA_ERROR"); + if (!p.get("collapse_key").isNull()) identifier(p.get("collapse_key")); + if (!p.get("agg_sort").isNull()) { + JsonNode sort = p.get("agg_sort"); + exactKeys(sort, "field", "operator", "ascending"); + identifier(sort.get("field")); + identifier(sort.get("operator")); + if (!sort.get("ascending").isBoolean()) throw failure("CONFIG_SCHEMA_ERROR"); + } + return (ObjectNode) n; + } catch (AdapterException e) { throw e; } + catch (Exception e) { throw failure("CONFIG_SCHEMA_ERROR"); } + } + + private void exactKeys(JsonNode node, String... keys) { + if (node == null || !node.isObject() || node.size() != keys.length) + throw failure("CONFIG_SCHEMA_ERROR"); + for (String key : keys) if (!node.has(key)) throw failure("CONFIG_SCHEMA_ERROR"); + } + + private void identifier(JsonNode node) { + if (!node.isTextual() || !node.textValue().matches("[A-Za-z0-9_.-]{0,128}")) + throw failure("CONFIG_SCHEMA_ERROR"); + } + + @Override public PullPage pull(PullRequest r) { + DeviceCollectApiConfig c = r.getConfig(); + validateConfig(c); + ObjectNode options = options(c); + if (r.getLimit() < 1 || r.getLimit() > c.getPageSize() || r.getOffset() < 0 + || r.getOffset() > Integer.MAX_VALUE || c.getDeviceId() == null || c.getCollectTaskId() == null + || r.getQueryStart() == null || r.getQueryEnd() == null + || !r.getQueryEnd().isAfter(r.getQueryStart()) + || r.getQueryStart().getNano() % 1000000 != 0 || r.getQueryEnd().getNano() % 1000000 != 0) + throw failure("REQUEST_RANGE_INVALID"); + long start = r.getQueryStart().toEpochMilli(); + long end = r.getQueryEnd().toEpochMilli(); + String id = UUID.randomUUID().toString(); + ObjectNode body = json.createObjectNode(); + body.put("jsonrpc", "2.0"); + body.put("method", "AlarmService.GetAlarmDocument"); + body.put("id", id); + ObjectNode params = body.putObject("params"); + params.put("offset", r.getOffset()); + params.put("count", r.getLimit()); + params.put("time_range_start", start); + params.put("time_range_end", options.get("endInclusive").booleanValue() ? end - 1 : end); + options.get("queryParams").fields().forEachRemaining(e -> { + if (!e.getValue().isNull()) params.set(e.getKey(), e.getValue()); + }); + HttpPost post = new HttpPost(urls.validate(c.getBaseUrl())); + post.setConfig(RequestConfig.custom().setConnectTimeout(c.getConnectTimeoutMs()) + .setConnectionRequestTimeout(c.getConnectTimeoutMs()).setSocketTimeout(c.getRequestTimeoutMs()) + .setRedirectsEnabled(false).build()); + post.setHeader("Accept", "application/json"); + byte[] responseBytes = null; + long started = System.nanoTime(); + try (CredentialSnapshot credential = credentials.load(c.getCredentialRef(), TYPE, c.getDeviceId())) { + char[] token = credential.require("api_token").copy(); + try { + CredentialSchemas.validate(TYPE, null, null, new String(token)); + post.setHeader("API-Token", new String(token)); + } finally { Arrays.fill(token, '\0'); } + post.setEntity(new StringEntity(json.writeValueAsString(body), ContentType.APPLICATION_JSON)); + try (CloseableHttpResponse response = http.execute(post)) { + int status = response.getStatusLine().getStatusCode(); + // No remote error body/message (which may echo a credential) is ever propagated. + if (status < 200 || status >= 300) { + String code = status == 401 ? "AUTH_HTTP_401" : status == 403 ? "AUTH_HTTP_403" : "HTTP_ERROR"; + throw new AdapterException(code, "全悉 HTTP 请求失败: " + status, status); + } + responseBytes = readBounded(response.getEntity(), c.getMaxResponseBytes()); + long httpMs = elapsed(started); + long parsing = System.nanoTime(); + JsonNode root = parseBounded(responseBytes); + if (!root.isObject() || !"2.0".equals(root.path("jsonrpc").asText()) + || !root.path("id").isTextual() || !id.equals(root.path("id").textValue())) + throw failure("RPC_ENVELOPE_INVALID"); + if (root.has("error")) throw failure("RPC_ERROR"); + JsonNode result = root.get("result"); + if (result == null || !result.isObject()) throw failure("RESPONSE_SCHEMA_ERROR"); + long total = count(result.get("total")); + long pageTotal = count(result.get("page_total")); + long selectedTotal = "total".equals(options.get("paginationTotalField").asText()) ? total : pageTotal; + JsonNode data = result.get("data"); + if (data == null || !data.isArray() || data.size() > r.getLimit() + || r.getOffset() + data.size() > selectedTotal + || (data.size() == 0 && r.getOffset() < selectedTotal)) + throw failure("PAGINATION_INCONSISTENT"); + List events = new ArrayList<>(); + Set ids = new HashSet<>(); + for (JsonNode item : data) { + if (!item.isObject()) throw failure("RESPONSE_SCHEMA_ERROR"); + String external = requiredText(item.get("doc_id"), 512); + if (!ids.add(external)) throw failure("PAGINATION_DUPLICATE_DOCUMENT"); + JsonNode time = item.get("timestamp"); + if (time == null || !time.isIntegralNumber() || !time.canConvertToLong() + || time.longValue() < 946684800000L || time.longValue() >= 4102444800000L) + throw failure("EVENT_TIME_INVALID"); + Instant at = Instant.ofEpochMilli(time.longValue()); + if (at.isBefore(r.getQueryStart()) || !at.isBefore(r.getQueryEnd())) + throw failure("EVENT_OUTSIDE_WINDOW"); + String raw = json.writeValueAsString(item); + if (raw.getBytes(StandardCharsets.UTF_8).length > 1048576) throw failure("EVENT_TOO_LARGE"); + // Canonical object order, original array order, all vendor fields including collapse_info. + // A changed document becomes a new revision; a byte-equivalent snapshot is idempotent. + String revision = hex(MessageDigest.getInstance("SHA-256") + .digest(json.writeValueAsBytes(canonical(item)))); + UUID task = uuidV5(NAMESPACE, c.getDeviceId() + "|" + c.getCollectTaskId() + "|" + TYPE + "|" + STREAM); + String eventId = ID_DATE.format(at) + "-" + uuidV5(task, external + "|" + revision); + events.add(new PulledEvent(eventId, external, at, raw)); + } + return new PullPage(events, selectedTotal, status, responseBytes.length, httpMs, + elapsed(parsing), total + ":" + pageTotal); + } + } catch (AdapterException | CredentialException e) { throw e; } + catch (SocketTimeoutException | org.apache.http.conn.ConnectTimeoutException e) { + throw failure("HTTP_TIMEOUT"); + } catch (IOException e) { throw failure("DEVICE_API_UNAVAILABLE"); } + catch (Exception e) { throw failure("RESPONSE_SCHEMA_ERROR"); } + finally { + post.removeHeaders("API-Token"); + post.releaseConnection(); + if (responseBytes != null) Arrays.fill(responseBytes, (byte) 0); + } + } + + private JsonNode parseBounded(byte[] bytes) { + try { + // Enforce depth before building a tree, also on unknown fields. + try (JsonParser p = json.getFactory().createParser(bytes)) { + int depth = 0; + int tokens = 0; + JsonToken t; + while ((t = p.nextToken()) != null) { + if (++tokens > 2000000) throw failure("RESPONSE_SCHEMA_ERROR"); + if (t == JsonToken.START_OBJECT || t == JsonToken.START_ARRAY) { + if (++depth > 64) throw failure("RESPONSE_SCHEMA_ERROR"); + } else if (t == JsonToken.END_OBJECT || t == JsonToken.END_ARRAY) depth--; + else if ((t == JsonToken.VALUE_STRING && p.getTextLength() > 1048576) + || (t == JsonToken.FIELD_NAME && p.getCurrentName().length() > 512)) + throw failure("RESPONSE_SCHEMA_ERROR"); + } + } + try (JsonParser p = json.getFactory().createParser(bytes)) { + JsonNode value = json.readTree(p); + if (value == null || p.nextToken() != null) throw failure("RESPONSE_SCHEMA_ERROR"); + return value; + } + } catch (AdapterException e) { throw e; } + catch (Exception e) { throw failure("RESPONSE_SCHEMA_ERROR"); } + } + + private byte[] readBounded(HttpEntity entity, long max) throws IOException { + if (entity == null) throw failure("RESPONSE_SCHEMA_ERROR"); + if (entity.getContentLength() > max) throw failure("RESPONSE_TOO_LARGE"); + try (InputStream in = entity.getContent(); ByteArrayOutputStream out = new ByteArrayOutputStream()) { + byte[] buffer = new byte[8192]; + int n; + long size = 0; + while ((n = in.read(buffer)) != -1) { + size += n; + if (size > max) throw failure("RESPONSE_TOO_LARGE"); + out.write(buffer, 0, n); + } + return out.toByteArray(); + } + } + + private JsonNode canonical(JsonNode value) { + if (value.isObject()) { + ObjectNode sorted = json.createObjectNode(); + TreeSet names = new TreeSet<>(); + value.fieldNames().forEachRemaining(names::add); + for (String name : names) sorted.set(name, canonical(value.get(name))); + return sorted; + } + if (value.isArray()) { + com.fasterxml.jackson.databind.node.ArrayNode array = json.createArrayNode(); + value.forEach(v -> array.add(canonical(v))); + return array; + } + return value; + } + + private long count(JsonNode value) { + if (value == null || !value.isIntegralNumber() || !value.canConvertToLong() || value.longValue() < 0) + throw failure("RESPONSE_SCHEMA_ERROR"); + return value.longValue(); + } + private String requiredText(JsonNode value, int max) { + if (value == null || !value.isTextual() || value.textValue().trim().isEmpty() + || value.textValue().length() > max) throw failure("RESPONSE_SCHEMA_ERROR"); + return value.textValue(); + } + private void range(Integer value, int min, int max) { + if (value == null || value < min || value > max) throw failure("CONFIG_SCHEMA_ERROR"); + } + private static String hex(byte[] bytes) { + StringBuilder out = new StringBuilder(bytes.length * 2); + for (byte b : bytes) out.append(String.format("%02x", b & 255)); + return out.toString(); + } + private static UUID uuidV5(UUID ns, String value) throws Exception { + MessageDigest digest = MessageDigest.getInstance("SHA-1"); + digest.update(ByteBuffer.allocate(16).putLong(ns.getMostSignificantBits()).putLong(ns.getLeastSignificantBits()).array()); + byte[] hash = digest.digest(value.getBytes(StandardCharsets.UTF_8)); + hash[6] = (byte) ((hash[6] & 15) | 80); + hash[8] = (byte) ((hash[8] & 63) | 128); + ByteBuffer b = ByteBuffer.wrap(hash); + return new UUID(b.getLong(), b.getLong()); + } + private long elapsed(long start) { return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); } + private AdapterException failure(String code) { return new AdapterException(code, "全悉 API: " + code); } + @PreDestroy public void close() { try { http.close(); } catch (IOException ignored) { } } +} 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 bdb62ff..42c326d 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 @@ -212,13 +212,20 @@ public final class ApiPullDtos { @Pattern(regexp = "^[A-Za-z0-9._-]{1,64}$") private String credentialVersion; - @NotBlank + @lombok.ToString.Exclude + @com.fasterxml.jackson.annotation.JsonProperty(access = com.fasterxml.jackson.annotation.JsonProperty.Access.WRITE_ONLY) @Size(max = 512) private String appKey; - @NotBlank + @lombok.ToString.Exclude + @com.fasterxml.jackson.annotation.JsonProperty(access = com.fasterxml.jackson.annotation.JsonProperty.Access.WRITE_ONLY) @Size(max = 2048) private String appSecret; + + @lombok.ToString.Exclude + @com.fasterxml.jackson.annotation.JsonProperty(access = com.fasterxml.jackson.annotation.JsonProperty.Access.WRITE_ONLY) + @Size(max = 2048) + private String apiToken; } /** 轮换凭据时必须携带当前 configVersion，设备、适配器和引用不可原地修改。 */ @@ -231,13 +238,20 @@ public final class ApiPullDtos { @Pattern(regexp = "^[A-Za-z0-9._-]{1,64}$") private String credentialVersion; - @NotBlank + @lombok.ToString.Exclude + @com.fasterxml.jackson.annotation.JsonProperty(access = com.fasterxml.jackson.annotation.JsonProperty.Access.WRITE_ONLY) @Size(max = 512) private String appKey; - @NotBlank + @lombok.ToString.Exclude + @com.fasterxml.jackson.annotation.JsonProperty(access = com.fasterxml.jackson.annotation.JsonProperty.Access.WRITE_ONLY) @Size(max = 2048) private String appSecret; + + @lombok.ToString.Exclude + @com.fasterxml.jackson.annotation.JsonProperty(access = com.fasterxml.jackson.annotation.JsonProperty.Access.WRITE_ONLY) + @Size(max = 2048) + private String apiToken; } /** 凭据管理响应；刻意不定义 appKey/appSecret 字段，避免序列化或日志误回显。 */ diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/CredentialSchemas.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/CredentialSchemas.java new file mode 100644 index 0000000..1fa8448 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/CredentialSchemas.java @@ -0,0 +1,28 @@ +package com.haobang.apipull.credential; + +/** Fixed vendor schemas; never infer authentication from whichever secret happens to be present. */ +public final class CredentialSchemas { + private CredentialSchemas() {} + + public static void validate(String type, String key, String secret, String token) { + if ("NDRSERVER_V1".equals(type)) { + require(key, 512); + require(secret, 2048); + if (token != null) fail(); + } else if ("QUANXI_V1".equals(type)) { + require(token, 2048); + if (key != null || secret != null) fail(); + // Header value must be printable ASCII, with no CR/LF or surrounding whitespace. + for (int i = 0; i < token.length(); i++) + if (token.charAt(i) < 33 || token.charAt(i) > 126) fail(); + } else fail(); + } + + private static void require(String value, int max) { + if (value == null || value.trim().isEmpty() || value.length() > max) fail(); + } + + private static void fail() { + throw new CredentialException("CREDENTIAL_SCHEMA", "凭据字段不符合厂商认证 Schema"); + } +} diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/DatabaseCredentialProvider.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/DatabaseCredentialProvider.java index 89e2358..bec4cfe 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/DatabaseCredentialProvider.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/credential/DatabaseCredentialProvider.java @@ -32,15 +32,18 @@ public class DatabaseCredentialProvider implements CredentialProvider { if (credential == null) { throw new CredentialException("CREDENTIAL_NOT_FOUND", "设备凭据不存在或绑定关系不匹配"); } - if (isBlank(credential.getCredentialVersion()) - || isBlank(credential.getAppKey()) - || isBlank(credential.getAppSecret())) { - throw new CredentialException("CREDENTIAL_SCHEMA", "设备凭据字段不完整"); + if (isBlank(credential.getCredentialVersion())) { + throw new CredentialException("CREDENTIAL_SCHEMA", "设备凭据版本缺失"); } - + CredentialSchemas.validate(expectedAdapterType, credential.getAppKey(), + credential.getAppSecret(), credential.getApiToken()); Map values = new LinkedHashMap<>(); - values.put("app_key", new SecretValue(credential.getAppKey().toCharArray())); - values.put("app_secret", new SecretValue(credential.getAppSecret().toCharArray())); + if ("QUANXI_V1".equals(expectedAdapterType)) { + values.put("api_token", new SecretValue(credential.getApiToken().toCharArray())); + } else { + values.put("app_key", new SecretValue(credential.getAppKey().toCharArray())); + values.put("app_secret", new SecretValue(credential.getAppSecret().toCharArray())); + } return new CredentialSnapshot( 1, credential.getCredentialVersion(), credential.getAdapterType(), values); } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/persistence/DeviceApiCredential.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/persistence/DeviceApiCredential.java index 0b03ab4..8b80263 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/persistence/DeviceApiCredential.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/persistence/DeviceApiCredential.java @@ -15,6 +15,10 @@ public class DeviceApiCredential { private String credentialVersion; private String appKey; private String appSecret; + private String apiToken; + + public String getApiToken() { return apiToken; } + public void setApiToken(String apiToken) { this.apiToken = apiToken; } private long configVersion; private LocalDateTime createdAt; private LocalDateTime updatedAt; diff --git a/haobang-security-dm/syslog-serve/src/main/resources/application.properties b/haobang-security-dm/syslog-serve/src/main/resources/application.properties index b20a276..18ff490 100644 --- a/haobang-security-dm/syslog-serve/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-serve/src/main/resources/application.properties @@ -105,7 +105,7 @@ mybatis.configuration.map-underscore-to-camel-case=true device.api.enabled=${DEVICE_API_ENABLED:true} # 设备 AppKey/AppSecret 由 device_collect_api_credential 表管理，配置文件不再保存厂商凭据。 device.api.scheduler.instance-id=${DEVICE_API_INSTANCE_ID:${HOSTNAME:local}} -device.api.scheduler.scan-interval-ms=${DEVICE_API_SCAN_INTERVAL_MS:5000} +device.api.scheduler.scan-interval-ms=${DEVICE_API_SCAN_INTERVAL_MS:1000} device.api.scheduler.lease-seconds=${DEVICE_API_LEASE_SECONDS:120} # 同一任务页内严格串行；不同设备可并发。同一设备主机默认仅一个在途请求。 device.api.scheduler.global-concurrency=${DEVICE_API_GLOBAL_CONCURRENCY:4} diff --git a/haobang-security-dm/syslog-serve/src/main/resources/logback.xml b/haobang-security-dm/syslog-serve/src/main/resources/logback.xml index f90779a..d282e01 100644 --- a/haobang-security-dm/syslog-serve/src/main/resources/logback.xml +++ b/haobang-security-dm/syslog-serve/src/main/resources/logback.xml @@ -30,6 +30,11 @@ + + + + + diff --git a/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceApiCredentialMapper.xml b/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceApiCredentialMapper.xml index 4b57cd8..19c1d75 100644 --- a/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceApiCredentialMapper.xml +++ b/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceApiCredentialMapper.xml @@ -4,9 +4,11 @@ INSERT INTO device_collect_api_credential (credential_ref, device_id, adapter_type, credential_version, app_key, app_secret, + api_token, config_version, created_at, updated_at) VALUES (#{credentialRef}, #{deviceId}, #{adapterType}, #{credentialVersion}, #{appKey}, #{appSecret}, + #{apiToken}, 1, SYSDATE, SYSDATE) @@ -19,6 +21,7 @@