feat(api-pull): 完善全悉设备接入并修复 NDR 边界重复采集
- 增加全悉设备 API 日志拉取及消费处理 - 完善全悉日志解析、字段映射和标准化逻辑 - NDR 采集窗口统一采用左闭右开规则 - 对设备秒级查询范围向外取整，并在本地精确过滤 - 区分接口原始返回数和实际接收数，保证分页偏移正确推进 - 补充时间窗口、边界过滤和分页推进测试 - 更新设备 API 拉取及对接说明文档
This commit is contained in:
王绍宏
+5
@@ -944,6 +944,11 @@ public class LogNormalProcessor {
|
||||
(HashMap<String, Object>) ((HashMap<String, Object>) map.get("action")).get("param");
|
||||
normalColumMap.put("dest_field_value",
|
||||
getMappingValue(action_param, entry.getValue().toString()));
|
||||
} else if (((HashMap<String, Object>) map.get("action")).get("type").equals("bounded_text")) {
|
||||
Map<String, Object> params = (Map<String, Object>)
|
||||
((Map<String, Object>) 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<String, Object>) map.get("action")).get("type").equals("time")) {
|
||||
normalColumMap.put("action_param",
|
||||
((HashMap<String, Object>) map.get("action")).get("param"));
|
||||
|
||||
+4
-1
@@ -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);
|
||||
|
||||
+7
@@ -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,
|
||||
|
||||
+32
@@ -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;
|
||||
}
|
||||
}
|
||||
+165
@@ -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<String, Object> 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<n;i++) b.append(s); return b.toString(); }
|
||||
static Path file(String name) {
|
||||
Path root = Paths.get("").toAbsolutePath();
|
||||
while (!Files.isDirectory(root.resolve("design/device-integration-docs/quanxi"))) root = root.getParent();
|
||||
return root.resolve("design/device-integration-docs/quanxi/" + name).normalize();
|
||||
}
|
||||
static class Fixture {
|
||||
final DeviceCollectTaskService tasks = mock(DeviceCollectTaskService.class);
|
||||
final DeviceDeviceService devices = mock(DeviceDeviceService.class);
|
||||
final SyslogNormalAlarmMapper alarm = mock(SyslogNormalAlarmMapper.class);
|
||||
final SyslogNormalDataMapper normal = mock(SyslogNormalDataMapper.class);
|
||||
final SyslogNormalDataService store = new SyslogNormalDataService();
|
||||
final LogNormalProcessor processor;
|
||||
final BatchRuleSnapshot snapshot;
|
||||
Fixture(boolean compiled) throws Exception {
|
||||
ReflectionTestUtils.setField(store, "syslogNormalAlarmMapper", alarm);
|
||||
ReflectionTestUtils.setField(store, "syslogNormalDataMapper", normal);
|
||||
ReflectionTestUtils.setField(store, "statsChangeCaptureService", mock(StatsChangeCaptureService.class));
|
||||
ReflectionTestUtils.setField(store, "directStatsProjectionWriter", mock(DirectStatsProjectionWriter.class));
|
||||
when(alarm.insertDynamic(anyMap())).thenReturn(1);
|
||||
NormalizationOptimizationProperties p = new NormalizationOptimizationProperties();
|
||||
p.getCompiledRule().setEnabled(compiled); p.getCompiledRule().setShadowCompareEnabled(false);
|
||||
p.getRuleSelection().setEnabled(true); p.getRuleSelection().setShadowEnabled(false);
|
||||
CompiledRuleCompiler compiler = new CompiledRuleCompiler(JSON);
|
||||
processor = new LogNormalProcessor(mock(DmNormalizeRuleService.class), mock(DmColumnService.class), store,
|
||||
mock(SyslogNonNormalMessageService.class), new LogDataFilterService(), new LogDataCompleteService(),
|
||||
tasks, devices, mock(DmColumnMapper.class), mock(DmNormalizeRuleMapper.class),
|
||||
mock(KafkaConsumerProperties.class), mock(AppProperties.class), new CompiledRuleCache(compiler, p), p);
|
||||
String content = new String(Files.readAllBytes(file("normalization-rule.json")), "UTF-8");
|
||||
RuleSource source = new RuleSource(700L, "quanxi", "quanxi", "json", content, true, "0");
|
||||
snapshot = new BatchRuleSnapshot(Collections.singletonMap(7L, Collections.singletonList(source)), 1L);
|
||||
}
|
||||
Map<String, Object> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
+8
-4
@@ -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());
|
||||
|
||||
+28
-8
@@ -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<String> 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<CompletionStage<com.haobang.wal.DurableReceipt>> 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(),
|
||||
|
||||
+17
-7
@@ -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";
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+3
@@ -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);
|
||||
|
||||
|
||||
+62
-1
@@ -9,7 +9,9 @@ import java.util.List;
|
||||
* <p>{@code total} 是当前固定查询窗口的数据集总数；执行器要求后续各页 total 保持一致， 防止 offset 分页过程中数据集漂移而静默漏数。耗时和字节数用于慢页日志。
|
||||
*/
|
||||
public final class PullPage {
|
||||
private final String datasetFingerprint;
|
||||
private final List<PulledEvent> 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<PulledEvent> events,
|
||||
long total,
|
||||
int httpStatus,
|
||||
long responseBytes,
|
||||
long httpDurationMs,
|
||||
long parseDurationMs,
|
||||
String datasetFingerprint) {
|
||||
this(
|
||||
events,
|
||||
total,
|
||||
httpStatus,
|
||||
responseBytes,
|
||||
httpDurationMs,
|
||||
parseDurationMs,
|
||||
datasetFingerprint,
|
||||
events.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建允许适配器过滤厂商边界记录的结果页。
|
||||
*
|
||||
* <p>{@code sourceRecordCount} 是设备本页实际返回的元素数，{@code events} 是适配器按通用逻辑窗口
|
||||
* 过滤后允许进入 WAL 的元素。Runner 必须按前者推进 offset，按后者写入 WAL；否则边界页被全部过滤时
|
||||
* 会重复请求同一页，或者错误地把设备响应判断为空页。
|
||||
*/
|
||||
public PullPage(
|
||||
List<PulledEvent> 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<PulledEvent> getEvents() {
|
||||
return events;
|
||||
}
|
||||
|
||||
/** 设备本页实际返回的元素数，用于正确推进厂商 offset。 */
|
||||
public int getSourceRecordCount() {
|
||||
return sourceRecordCount;
|
||||
}
|
||||
|
||||
public long getTotal() {
|
||||
return total;
|
||||
}
|
||||
|
||||
+30
-3
@@ -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<PulledEvent> 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())) {
|
||||
|
||||
+320
@@ -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<String> 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<PulledEvent> events = new ArrayList<>();
|
||||
Set<String> 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<String> 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) { } }
|
||||
}
|
||||
+18
-4
@@ -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 字段，避免序列化或日志误回显。 */
|
||||
|
||||
+28
@@ -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");
|
||||
}
|
||||
}
|
||||
+8
-5
@@ -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<String, SecretValue> values = new LinkedHashMap<>();
|
||||
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);
|
||||
}
|
||||
|
||||
+4
@@ -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;
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -30,6 +30,11 @@
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- HTTP wire/header 与凭据 SQL 包含原始密钥，禁止随 root DEBUG 打开。 -->
|
||||
<logger name="org.apache.http.wire" level="OFF"/>
|
||||
<logger name="org.apache.http.headers" level="OFF"/>
|
||||
<logger name="com.common.mapper.DeviceApiCredentialMapper" level="OFF"/>
|
||||
|
||||
<!-- 设置日志级别 -->
|
||||
<root level="INFO">
|
||||
<appender-ref ref="CONSOLE" />
|
||||
|
||||
+4
@@ -4,9 +4,11 @@
|
||||
<insert id="insert">
|
||||
INSERT INTO device_collect_api_credential
|
||||
(credential_ref, device_id, adapter_type, credential_version, app_key, app_secret,
|
||||
<if test="adapterType == 'QUANXI_V1'">api_token,</if>
|
||||
config_version, created_at, updated_at)
|
||||
VALUES
|
||||
(#{credentialRef}, #{deviceId}, #{adapterType}, #{credentialVersion}, #{appKey}, #{appSecret},
|
||||
<if test="adapterType == 'QUANXI_V1'">#{apiToken},</if>
|
||||
1, SYSDATE, SYSDATE)
|
||||
</insert>
|
||||
|
||||
@@ -19,6 +21,7 @@
|
||||
|
||||
<select id="selectForLoad" resultType="com.haobang.apipull.persistence.DeviceApiCredential">
|
||||
SELECT credential_ref, device_id, adapter_type, credential_version, app_key, app_secret,
|
||||
<if test="adapterType == 'QUANXI_V1'">api_token,</if>
|
||||
config_version, created_at, updated_at
|
||||
FROM device_collect_api_credential
|
||||
WHERE credential_ref = #{credentialRef}
|
||||
@@ -31,6 +34,7 @@
|
||||
SET credential_version = #{credentialVersion},
|
||||
app_key = #{appKey},
|
||||
app_secret = #{appSecret},
|
||||
<if test="adapterType == 'QUANXI_V1'">api_token = #{apiToken},</if>
|
||||
config_version = config_version + 1,
|
||||
updated_at = SYSDATE
|
||||
WHERE credential_ref = #{credentialRef}
|
||||
|
||||
+86
@@ -212,6 +212,92 @@ class ApiPullRunnerTest {
|
||||
any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void duplicateDocumentsOrChangedMetadataNeverCommit() {
|
||||
for (boolean duplicate : new boolean[]{true, false}) {
|
||||
RunnerFixture f = new RunnerFixture();
|
||||
f.config.setPageSize(1);
|
||||
when(f.adapter.requiresDistinctExternalIds()).thenReturn(true);
|
||||
PullPage first = new PullPage(page(2, 1, 0).getEvents(), 2, 200, 20, 0, 0, "2:2");
|
||||
PullPage second = new PullPage(page(2, 1, duplicate ? 0 : 1).getEvents(), 2, 200, 20, 0, 0,
|
||||
duplicate ? "2:2" : "3:2");
|
||||
when(f.adapter.pull(any())).thenReturn(first, second);
|
||||
f.runner.execute(f.config, f.checkpoint, f.run);
|
||||
assertEquals("FAILED", f.run.getStatus());
|
||||
assertEquals(duplicate ? "PAGINATION_DUPLICATE_DOCUMENT" : "PAGINATION_DATASET_CHANGED", f.run.getErrorCode());
|
||||
verify(f.mapper, never()).commitCheckpoint(anyInt(), anyLong(), anyString(), anyLong(),
|
||||
anyLong(), anyString(), any(), any(), any());
|
||||
verify(f.ingress, times(1)).tryAppend(any());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void secondPageFailureRestartsSameWindowAtOffsetZero() {
|
||||
RunnerFixture f = new RunnerFixture(); f.config.setPageSize(1);
|
||||
when(f.adapter.pull(any())).thenReturn(page(2, 1, 0))
|
||||
.thenThrow(new com.haobang.apipull.adapter.AdapterException("HTTP_TIMEOUT", "offline timeout"))
|
||||
.thenReturn(page(2, 1, 0), page(2, 1, 1));
|
||||
f.runner.execute(f.config, f.checkpoint, f.run);
|
||||
assertEquals("FAILED", f.run.getStatus());
|
||||
verify(f.mapper, never()).commitCheckpoint(anyInt(), anyLong(), anyString(), anyLong(),
|
||||
anyLong(), anyString(), any(), any(), any());
|
||||
f.runner.execute(f.config, f.checkpoint, run());
|
||||
org.mockito.ArgumentCaptor<PullRequest> requests = org.mockito.ArgumentCaptor.forClass(PullRequest.class);
|
||||
verify(f.adapter, times(4)).pull(requests.capture());
|
||||
assertEquals(0, requests.getAllValues().get(2).getOffset());
|
||||
assertEquals(requests.getAllValues().get(0).getQueryStart(), requests.getAllValues().get(2).getQueryStart());
|
||||
assertEquals(requests.getAllValues().get(0).getQueryEnd(), requests.getAllValues().get(2).getQueryEnd());
|
||||
verify(f.mapper).commitCheckpoint(anyInt(), anyLong(), anyString(), anyLong(), anyLong(), anyString(), any(), any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void failedDurabilityEmptyEarlyPageAndTotalDriftDoNotCommit() {
|
||||
for (int scenario = 0; scenario < 3; scenario++) {
|
||||
RunnerFixture f = new RunnerFixture(); f.config.setPageSize(1);
|
||||
if (scenario == 0) {
|
||||
CompletableFuture<DurableReceipt> failed = new CompletableFuture<>();
|
||||
failed.completeExceptionally(new java.io.IOException("offline force failure"));
|
||||
when(f.ingress.tryAppend(any())).thenReturn(AppendAttempt.accepted(failed));
|
||||
}
|
||||
when(f.adapter.pull(any())).thenReturn(page(2, 1, 0),
|
||||
scenario == 1 ? page(2, 0, 1) : page(3, 1, 1));
|
||||
f.runner.execute(f.config, f.checkpoint, f.run);
|
||||
assertEquals("FAILED", f.run.getStatus());
|
||||
verify(f.mapper, never()).commitCheckpoint(anyInt(), anyLong(), anyString(), anyLong(),
|
||||
anyLong(), anyString(), any(), any(), any());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void filteredBoundaryPageAdvancesVendorOffsetWithoutWritingWal() {
|
||||
RunnerFixture fixture = new RunnerFixture();
|
||||
fixture.config.setPageSize(1);
|
||||
PullPage filteredBoundary =
|
||||
new PullPage(
|
||||
java.util.Collections.emptyList(),
|
||||
2,
|
||||
200,
|
||||
20L,
|
||||
10L,
|
||||
1L,
|
||||
null,
|
||||
1);
|
||||
when(fixture.adapter.pull(any()))
|
||||
.thenReturn(filteredBoundary, page(2L, 1, 1L));
|
||||
|
||||
fixture.runner.execute(fixture.config, fixture.checkpoint, fixture.run);
|
||||
|
||||
assertEquals("SUCCESS", fixture.run.getStatus());
|
||||
assertEquals(Long.valueOf(1L), fixture.run.getRecordsFetched());
|
||||
assertEquals(Long.valueOf(1L), fixture.run.getRecordsDurable());
|
||||
verify(fixture.ingress, times(1)).tryAppend(any(IngressEvent.class));
|
||||
org.mockito.ArgumentCaptor<PullRequest> requests =
|
||||
org.mockito.ArgumentCaptor.forClass(PullRequest.class);
|
||||
verify(fixture.adapter, times(2)).pull(requests.capture());
|
||||
assertEquals(0L, requests.getAllValues().get(0).getOffset());
|
||||
assertEquals(1L, requests.getAllValues().get(1).getOffset());
|
||||
}
|
||||
|
||||
private static PullPage page(long total, int count, long offset) {
|
||||
List<PulledEvent> events = new ArrayList<>();
|
||||
for (int index = 0; index < count; index++) {
|
||||
|
||||
+99
@@ -227,6 +227,93 @@ class NdrServerApiAdapterTest {
|
||||
assertEquals(500, new ObjectMapper().readTree(requestBody).path("limit").asInt());
|
||||
}
|
||||
|
||||
@Test
|
||||
void adjacentLogicalWindowsEmitTheirSharedBoundaryOnlyOnce() throws Exception {
|
||||
CloseableHttpClient http = mock(CloseableHttpClient.class);
|
||||
CloseableHttpResponse challenge =
|
||||
response(200, "{\"result\":1,\"data\":{\"challenge\":\"c\"}}");
|
||||
CloseableHttpResponse verified =
|
||||
response(200, "{\"result\":1,\"data\":{\"token\":\"t\"}}");
|
||||
String boundaryResponse =
|
||||
"{\"result\":1,\"data\":{\"alerts\":[{\"index\":\"i\",\"id\":\"boundary\","
|
||||
+ "\"timestamp\":\"2024-07-02 15:01:00\"}],\"total\":1}}";
|
||||
CloseableHttpResponse previousBusiness = response(200, boundaryResponse);
|
||||
CloseableHttpResponse nextBusiness = response(200, boundaryResponse);
|
||||
when(http.execute(any(HttpUriRequest.class)))
|
||||
.thenReturn(challenge, verified, previousBusiness, nextBusiness);
|
||||
NdrServerApiAdapter adapter = adapter(http);
|
||||
|
||||
PullPage previous =
|
||||
adapter.pull(
|
||||
new PullRequest(
|
||||
config(),
|
||||
Instant.parse("2024-07-02T07:00:00Z"),
|
||||
Instant.parse("2024-07-02T07:01:00Z"),
|
||||
100,
|
||||
0));
|
||||
PullPage next =
|
||||
adapter.pull(
|
||||
new PullRequest(
|
||||
config(),
|
||||
Instant.parse("2024-07-02T07:01:00Z"),
|
||||
Instant.parse("2024-07-02T07:02:00Z"),
|
||||
100,
|
||||
0));
|
||||
|
||||
assertEquals(1, previous.getSourceRecordCount());
|
||||
assertTrue(previous.getEvents().isEmpty(), "右端点不属于前一个半开窗口");
|
||||
assertEquals(1, next.getSourceRecordCount());
|
||||
assertEquals(1, next.getEvents().size(), "左端点属于后一个半开窗口");
|
||||
|
||||
org.mockito.ArgumentCaptor<HttpUriRequest> requests =
|
||||
org.mockito.ArgumentCaptor.forClass(HttpUriRequest.class);
|
||||
verify(http, times(4)).execute(requests.capture());
|
||||
JsonNode previousBody = requestBody(requests.getAllValues().get(2));
|
||||
JsonNode nextBody = requestBody(requests.getAllValues().get(3));
|
||||
assertEquals("2024-07-02 15:01:00", previousBody.path("end_time").asText());
|
||||
assertEquals("2024-07-02 15:01:00", nextBody.path("start_time").asText());
|
||||
}
|
||||
|
||||
@Test
|
||||
void fractionalLogicalWindowIsExpandedWithoutLosingEventsInsideTheWindow() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
ObjectNode data = mapper.createObjectNode();
|
||||
data.putArray("logs")
|
||||
.add(flowAt(mapper, "before", "2024-07-02T15:00:00.568+08:00"))
|
||||
.add(flowAt(mapper, "at-start", "2024-07-02T15:00:00.569+08:00"))
|
||||
.add(flowAt(mapper, "before-end", "2024-07-02T15:00:15.568999+08:00"))
|
||||
.add(flowAt(mapper, "at-end", "2024-07-02T15:00:15.569+08:00"));
|
||||
data.put("total", 4);
|
||||
ObjectNode root = mapper.createObjectNode();
|
||||
root.put("result", 1);
|
||||
root.set("data", data);
|
||||
CloseableHttpClient http = mock(CloseableHttpClient.class);
|
||||
CloseableHttpResponse challenge =
|
||||
response(200, "{\"result\":1,\"data\":{\"challenge\":\"c\"}}");
|
||||
CloseableHttpResponse verified =
|
||||
response(200, "{\"result\":1,\"data\":{\"token\":\"t\"}}");
|
||||
CloseableHttpResponse business = response(200, mapper.writeValueAsString(root));
|
||||
when(http.execute(any(HttpUriRequest.class)))
|
||||
.thenReturn(challenge, verified, business);
|
||||
DeviceCollectApiConfig config = config();
|
||||
config.setStreamKey(NdrServerApiAdapter.FLOW);
|
||||
Instant start = Instant.parse("2024-07-02T07:00:00.569Z");
|
||||
Instant end = Instant.parse("2024-07-02T07:00:15.569Z");
|
||||
|
||||
PullPage page = adapter(http).pull(new PullRequest(config, start, end, 100, 0));
|
||||
|
||||
assertEquals(4, page.getSourceRecordCount());
|
||||
assertEquals(2, page.getEvents().size());
|
||||
assertEquals("at-start", page.getEvents().get(0).getExternalEventId());
|
||||
assertEquals("before-end", page.getEvents().get(1).getExternalEventId());
|
||||
org.mockito.ArgumentCaptor<HttpUriRequest> requests =
|
||||
org.mockito.ArgumentCaptor.forClass(HttpUriRequest.class);
|
||||
verify(http, times(3)).execute(requests.capture());
|
||||
JsonNode body = requestBody(requests.getAllValues().get(2));
|
||||
assertEquals("2024-07-02 15:00:00", body.path("start_time").asText());
|
||||
assertEquals("2024-07-02 15:00:16", body.path("end_time").asText());
|
||||
}
|
||||
|
||||
@Test
|
||||
void alertHttpPayloadGainsTcpCompatibleAliasesAndKeepsVendorFields() throws Exception {
|
||||
ObjectMapper mapper = new ObjectMapper();
|
||||
@@ -472,6 +559,11 @@ class NdrServerApiAdapterTest {
|
||||
return new NdrServerApiAdapter(provider, new SafeBaseUrlValidator(p), new ObjectMapper(), http);
|
||||
}
|
||||
|
||||
private JsonNode requestBody(HttpUriRequest request) throws Exception {
|
||||
HttpEntityEnclosingRequest entityRequest = (HttpEntityEnclosingRequest) request;
|
||||
return new ObjectMapper().readTree(EntityUtils.toString(entityRequest.getEntity(), "UTF-8"));
|
||||
}
|
||||
|
||||
private ObjectNode flowLog(ObjectMapper mapper) {
|
||||
ObjectNode log = mapper.createObjectNode();
|
||||
log.put("_id", "flow-1");
|
||||
@@ -480,6 +572,13 @@ class NdrServerApiAdapterTest {
|
||||
return log;
|
||||
}
|
||||
|
||||
private ObjectNode flowAt(ObjectMapper mapper, String id, String timestamp) {
|
||||
ObjectNode log = mapper.createObjectNode();
|
||||
log.put("_id", id);
|
||||
log.put("timestamp", timestamp);
|
||||
return log;
|
||||
}
|
||||
|
||||
private ObjectNode alertLog(ObjectMapper mapper) {
|
||||
ObjectNode alert = mapper.createObjectNode();
|
||||
alert.put("index", "ndralert-20260904");
|
||||
|
||||
+269
@@ -0,0 +1,269 @@
|
||||
package com.haobang.apipull.adapter.quanxi;
|
||||
|
||||
import com.fasterxml.jackson.databind.*;
|
||||
import com.fasterxml.jackson.databind.node.*;
|
||||
import com.haobang.apipull.*;
|
||||
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.file.*;
|
||||
import java.time.*;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.apache.http.*;
|
||||
import org.apache.http.client.methods.*;
|
||||
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;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
|
||||
/** No sockets, Spring Boot, DB, Kafka or vendor credentials: all HTTP executions are mocked. */
|
||||
class QuanxiApiAdapterTest {
|
||||
static final ObjectMapper JSON = new ObjectMapper();
|
||||
static final Instant START = Instant.parse("2026-08-10T00:00:00Z");
|
||||
static final Instant END = START.plusSeconds(86400);
|
||||
static final String TOKEN = "synthetic-unit-test-token";
|
||||
|
||||
@Test void realFixtureBecomesTwentyIndependentOriginalDocumentsAndGoldenIds() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
PullPage page = f.pull();
|
||||
JsonNode expected = JSON.readTree(file("fixtures/expected-events.json").toFile());
|
||||
assertEquals(20, page.getEvents().size());
|
||||
assertEquals(2311, page.getTotal());
|
||||
assertEquals("2311:2311", page.getDatasetFingerprint());
|
||||
for (int i = 0; i < 20; i++) {
|
||||
PulledEvent event = page.getEvents().get(i);
|
||||
assertEquals(f.root.path("result").path("data").get(i), JSON.readTree(event.getRawMessage()));
|
||||
assertEquals(expected.get(i).path("event_id").asText(), event.getEventId());
|
||||
assertEquals(expected.get(i).path("event_time_ms").asLong(), event.getEventTime().toEpochMilli());
|
||||
assertEquals(expected.get(i).path("doc_id").asText(), event.getExternalEventId());
|
||||
assertFalse(JSON.readTree(event.getRawMessage()).has("http_req_body"));
|
||||
}
|
||||
assertEquals(page.getEvents().get(6).getEventTime(), page.getEvents().get(7).getEventTime());
|
||||
assertNotEquals(page.getEvents().get(6).getEventId(), page.getEvents().get(7).getEventId());
|
||||
}
|
||||
|
||||
@Test void protocolHeaderPaginationAndMillisecondsAreExact() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.pull(20);
|
||||
JsonNode body = f.request.get();
|
||||
assertEquals("AlarmService.GetAlarmDocument", body.path("method").asText());
|
||||
assertEquals("2.0", body.path("jsonrpc").asText());
|
||||
JsonNode p = body.path("params");
|
||||
assertEquals(20, p.path("offset").asLong());
|
||||
assertEquals(20, p.path("count").asInt());
|
||||
assertEquals(START.toEpochMilli(), p.path("time_range_start").asLong());
|
||||
assertEquals(END.toEpochMilli() - 1, p.path("time_range_end").asLong());
|
||||
assertFalse(p.has("ignore"));
|
||||
assertFalse(p.has("collapse_key"));
|
||||
assertFalse(p.has("agg_sort"));
|
||||
}
|
||||
|
||||
@Test void configuredParamsDoNotBecomeGuessedDefaults() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
ObjectNode opts = (ObjectNode) JSON.readTree(f.config.getExtraConfigJson());
|
||||
ObjectNode p = (ObjectNode) opts.get("queryParams");
|
||||
p.put("ignore", 0); p.put("collapse_key", "collapse_field");
|
||||
p.putObject("agg_sort").put("field", "maxtime").put("operator", "").put("ascending", false);
|
||||
opts.put("endInclusive", false); opts.put("paginationTotalField", "page_total");
|
||||
f.config.setExtraConfigJson(opts.toString());
|
||||
f.root.with("result").put("total", 9000).put("page_total", 2311);
|
||||
assertEquals(2311, f.pull().getTotal());
|
||||
assertEquals(END.toEpochMilli(), f.request.get().path("params").path("time_range_end").asLong());
|
||||
assertEquals(p.get("agg_sort"), f.request.get().path("params").get("agg_sort"));
|
||||
assertEquals(0, f.request.get().path("params").path("ignore").asInt());
|
||||
}
|
||||
|
||||
@Test void retriesAndObjectKeyOrderAreIdempotentButRevisionsAndNamespacesAreNot() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
String id = f.pull().getEvents().get(0).getEventId();
|
||||
assertEquals(id, f.pull().getEvents().get(0).getEventId());
|
||||
ObjectNode original = (ObjectNode) f.data().get(0);
|
||||
List<String> keys = new ArrayList<>(); original.fieldNames().forEachRemaining(keys::add);
|
||||
Collections.reverse(keys); ObjectNode reordered = JSON.createObjectNode();
|
||||
for (String key : keys) reordered.set(key, original.get(key));
|
||||
f.data().set(0, reordered);
|
||||
assertEquals(id, f.pull().getEvents().get(0).getEventId());
|
||||
reordered.with("collapse_info").put("count", "3");
|
||||
assertNotEquals(id, f.pull().getEvents().get(0).getEventId());
|
||||
f.data().set(0, original); f.config.setDeviceId(8);
|
||||
assertNotEquals(id, f.pull().getEvents().get(0).getEventId());
|
||||
f.config.setDeviceId(7); f.config.setCollectTaskId(12);
|
||||
assertNotEquals(id, f.pull().getEvents().get(0).getEventId());
|
||||
}
|
||||
|
||||
@Test void rpcErrorsAtHttp200NeverSucceedAndNeverEchoRemoteMessages() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
f.root.remove("result");
|
||||
f.root.putObject("error").put("code", -32001).put("message", TOKEN);
|
||||
AdapterException e = assertThrows(AdapterException.class, f::pull);
|
||||
assertEquals("RPC_ERROR", e.getCode()); assertFalse(e.toString().contains(TOKEN));
|
||||
assertNull(e.getCause());
|
||||
}
|
||||
|
||||
@Test void wrongVersionMissingOrMismatchedIdsAndMissingResultFail() throws Exception {
|
||||
for (String mode : Arrays.asList("version", "id", "result")) {
|
||||
Fixture f = new Fixture(); f.rewriteId = !mode.equals("id");
|
||||
if (mode.equals("version")) f.root.put("jsonrpc", "1.0");
|
||||
if (mode.equals("result")) f.root.remove("result");
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
}
|
||||
}
|
||||
|
||||
@Test void httpErrorsAreClassifiedWithoutRetryOrErrorBodyLeak() throws Exception {
|
||||
for (int status : new int[]{301, 401, 403, 429, 500}) {
|
||||
Fixture f = new Fixture(); f.status = status;
|
||||
AdapterException e = assertThrows(AdapterException.class, f::pull);
|
||||
assertEquals(Integer.valueOf(status), e.getHttpStatus());
|
||||
assertEquals(status == 401 ? "AUTH_HTTP_401" : status == 403 ? "AUTH_HTTP_403" : "HTTP_ERROR", e.getCode());
|
||||
verify(f.http, times(1)).execute(any(HttpUriRequest.class));
|
||||
}
|
||||
}
|
||||
|
||||
@Test void timeoutsDoNotLeakClientExceptions() throws Exception {
|
||||
Fixture f = new Fixture();
|
||||
when(f.http.execute(any(HttpUriRequest.class))).thenThrow(new SocketTimeoutException(TOKEN));
|
||||
AdapterException e = assertThrows(AdapterException.class, f::pull);
|
||||
assertEquals("HTTP_TIMEOUT", e.getCode()); assertFalse(e.toString().contains(TOKEN));
|
||||
assertNull(e.getCause());
|
||||
}
|
||||
|
||||
@Test void emptyArrayRequiresConsistentCountsAndExplicitData() throws Exception {
|
||||
Fixture f = new Fixture(); f.data().removeAll();
|
||||
assertEquals("PAGINATION_INCONSISTENT", assertThrows(AdapterException.class, f::pull).getCode());
|
||||
f.root.with("result").put("total", 0).put("page_total", 0);
|
||||
assertTrue(f.pull().getEvents().isEmpty());
|
||||
f.root.with("result").remove("data");
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
}
|
||||
|
||||
@Test void fractionalNegativeStringCountsAndOversizedPagesAreRejected() throws Exception {
|
||||
for (JsonNode value : Arrays.asList(JSON.readTree("1.5"), JSON.readTree("-1"), JSON.readTree("\"20\""))) {
|
||||
Fixture f = new Fixture(); f.root.with("result").set("total", value);
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
}
|
||||
Fixture f = new Fixture(); f.data().add(f.data().get(0));
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
}
|
||||
|
||||
@Test void sameDocTwiceCannotSatisfyPagination() throws Exception {
|
||||
Fixture f = new Fixture(); f.data().set(1, f.data().get(0));
|
||||
assertEquals("PAGINATION_DUPLICATE_DOCUMENT", assertThrows(AdapterException.class, f::pull).getCode());
|
||||
}
|
||||
|
||||
@Test void timestampIsMillisecondsNotSecondsMicrosOrFlowStart() throws Exception {
|
||||
for (long value : new long[]{1786377562L, 1786377562453000L, -1L}) {
|
||||
Fixture f = new Fixture(); ((ObjectNode) f.data().get(0)).put("timestamp", value);
|
||||
assertEquals("EVENT_TIME_INVALID", assertThrows(AdapterException.class, f::pull).getCode());
|
||||
}
|
||||
Fixture f = new Fixture();
|
||||
PulledEvent e = f.pull().getEvents().get(6);
|
||||
assertEquals(1786377127842L, e.getEventTime().toEpochMilli());
|
||||
assertNotEquals(f.data().get(6).path("flow_startts").asLong(), e.getEventTime().toEpochMilli());
|
||||
((ObjectNode) f.data().get(0)).put("timestamp", END.toEpochMilli());
|
||||
assertEquals("EVENT_OUTSIDE_WINDOW", assertThrows(AdapterException.class, f::pull).getCode());
|
||||
((ObjectNode) f.data().get(0)).put("timestamp", START.toEpochMilli());
|
||||
assertEquals(START, f.pull().getEvents().get(0).getEventTime());
|
||||
}
|
||||
|
||||
@Test void malformedJsonExcessiveDepthAndResponseBoundsFailClosed() throws Exception {
|
||||
for (String malformed : Arrays.asList("{", "{}{}", "{\"a\":1,\"a\":2}", nested(66))) {
|
||||
Fixture f = new Fixture(); f.rawOverride = malformed;
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
}
|
||||
Fixture f = new Fixture(); f.config.setMaxResponseBytes(1024L);
|
||||
assertEquals("RESPONSE_TOO_LARGE", assertThrows(AdapterException.class, f::pull).getCode());
|
||||
Fixture stream = new Fixture(); stream.unknownLength = true; stream.config.setMaxResponseBytes(1024L);
|
||||
assertEquals("RESPONSE_TOO_LARGE", assertThrows(AdapterException.class, stream::pull).getCode());
|
||||
}
|
||||
|
||||
@Test void unsafeOrUnconfiguredRequestsAreRejectedBeforeHttp() throws Exception {
|
||||
Fixture f = new Fixture(); f.config.setExtraConfigJson("{}");
|
||||
assertThrows(AdapterException.class, f::pull); verifyNoInteractions(f.http);
|
||||
f.config = config(); f.config.setBaseUrl("https://192.0.2.10/ndr/api");
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
f.config = config(); f.config.setBaseUrl("http://192.0.2.10/rpc");
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
f.config = config();
|
||||
assertThrows(AdapterException.class, () -> f.adapter.pull(new PullRequest(f.config, START, END, 20, -1)));
|
||||
assertThrows(AdapterException.class, () -> f.adapter.pull(new PullRequest(f.config, START.plusNanos(1), END, 20, 0)));
|
||||
verifyNoInteractions(f.http);
|
||||
}
|
||||
|
||||
@Test void existingFieldsAndVeryLargeBodiesAreNeverDuplicatedOrDecoded() throws Exception {
|
||||
Fixture f = new Fixture(); ObjectNode doc = (ObjectNode) f.data().get(6);
|
||||
doc.put("http_req_body", "already-present");
|
||||
doc.with("appbrief").with("http").put("req_body", repeat('x', 10001));
|
||||
JsonNode raw = JSON.readTree(f.pull().getEvents().get(6).getRawMessage());
|
||||
assertEquals(doc, raw);
|
||||
assertEquals("already-present", raw.path("http_req_body").asText());
|
||||
assertEquals(10001, raw.path("appbrief").path("http").path("req_body").asText().length());
|
||||
doc.put("payload", repeat('x', 1048577));
|
||||
assertThrows(AdapterException.class, f::pull);
|
||||
}
|
||||
|
||||
static String repeat(char c, int n) { char[] a = new char[n]; Arrays.fill(a, c); return new String(a); }
|
||||
static String nested(int n) { return repeat('[', n) + "0" + repeat(']', n); }
|
||||
static Path file(String name) {
|
||||
Path root = Paths.get("").toAbsolutePath();
|
||||
while (!Files.isDirectory(root.resolve("design/device-integration-docs/quanxi"))) root = root.getParent();
|
||||
return root.resolve("design/device-integration-docs/quanxi/" + name);
|
||||
}
|
||||
static DeviceCollectApiConfig config() {
|
||||
DeviceCollectApiConfig c = new DeviceCollectApiConfig();
|
||||
c.setDeviceId(7); c.setCollectTaskId(11); c.setAdapterType(QuanxiApiAdapter.TYPE);
|
||||
c.setStreamKey(QuanxiApiAdapter.STREAM); c.setBaseUrl("https://192.0.2.10/rpc");
|
||||
c.setCredentialRef("fixture"); c.setPageSize(20); c.setMaxPagesPerRun(100);
|
||||
c.setPollIntervalSeconds(10); c.setWindowSizeSeconds(60); c.setSafetyDelaySeconds(120);
|
||||
c.setLookbackSeconds(300); c.setConnectTimeoutMs(1000); c.setRequestTimeoutMs(1000);
|
||||
c.setMaxResponseBytes(16777216L);
|
||||
c.setExtraConfigJson("{\"recordMode\":\"DOCUMENT_SNAPSHOT\",\"paginationTotalField\":\"total\","
|
||||
+ "\"endInclusive\":true,\"queryParams\":{\"ignore\":null,\"collapse_key\":null,\"agg_sort\":null}}");
|
||||
return c;
|
||||
}
|
||||
static class Fixture {
|
||||
CloseableHttpClient http = mock(CloseableHttpClient.class);
|
||||
ObjectNode root = (ObjectNode) JSON.readTree(file("fixtures/api-response.json").toFile());
|
||||
AtomicReference<JsonNode> request = new AtomicReference<>();
|
||||
DeviceCollectApiConfig config = config();
|
||||
int status = 200; boolean rewriteId = true; boolean unknownLength;
|
||||
String rawOverride;
|
||||
QuanxiApiAdapter adapter;
|
||||
Fixture() throws Exception {
|
||||
ApiPullProperties p = new ApiPullProperties(); p.setAllowedHosts("192.0.2.10");
|
||||
CredentialProvider credentials = (ref, type, device) -> new CredentialSnapshot(1, "fixture", type,
|
||||
Collections.singletonMap("api_token", new SecretValue(TOKEN.toCharArray())));
|
||||
adapter = new QuanxiApiAdapter(credentials, new SafeBaseUrlValidator(p), JSON, http);
|
||||
when(http.execute(any(HttpUriRequest.class))).thenAnswer(inv -> {
|
||||
HttpPost post = inv.getArgument(0);
|
||||
assertEquals("https://192.0.2.10/rpc", post.getURI().toString());
|
||||
assertEquals(TOKEN, post.getFirstHeader("API-Token").getValue());
|
||||
assertNull(post.getFirstHeader("Authorization"));
|
||||
assertFalse(post.getConfig().isRedirectsEnabled());
|
||||
assertEquals(1000, post.getConfig().getSocketTimeout());
|
||||
request.set(JSON.readTree(EntityUtils.toString(post.getEntity(), "UTF-8")));
|
||||
ObjectNode envelope = root.deepCopy();
|
||||
if (rewriteId) envelope.set("id", request.get().get("id"));
|
||||
CloseableHttpResponse response = mock(CloseableHttpResponse.class);
|
||||
StatusLine line = mock(StatusLine.class); when(line.getStatusCode()).thenReturn(status);
|
||||
when(response.getStatusLine()).thenReturn(line);
|
||||
String text = rawOverride == null ? envelope.toString() : rawOverride;
|
||||
if (unknownLength) {
|
||||
HttpEntity entity = mock(HttpEntity.class); when(entity.getContentLength()).thenReturn(-1L);
|
||||
when(entity.getContent()).thenReturn(new ByteArrayInputStream(text.getBytes("UTF-8")));
|
||||
when(response.getEntity()).thenReturn(entity);
|
||||
} else when(response.getEntity()).thenReturn(new StringEntity(text, "UTF-8"));
|
||||
return response;
|
||||
});
|
||||
}
|
||||
ArrayNode data() { return (ArrayNode) root.path("result").path("data"); }
|
||||
PullPage pull() { return pull(0); }
|
||||
PullPage pull(long offset) { return adapter.pull(new PullRequest(config, START, END, 20, offset)); }
|
||||
}
|
||||
}
|
||||
+132
@@ -0,0 +1,132 @@
|
||||
package com.haobang.apipull.adapter.quanxi;
|
||||
|
||||
import com.common.entity.*;
|
||||
import com.common.mapper.*;
|
||||
import com.common.service.DeviceDeviceService;
|
||||
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.api.ApiPullDtos.*;
|
||||
import com.haobang.apipull.credential.*;
|
||||
import com.haobang.apipull.persistence.*;
|
||||
import com.haobang.apipull.security.SafeBaseUrlValidator;
|
||||
import com.haobang.config.AppProperties;
|
||||
import java.time.OffsetDateTime;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
import static org.mockito.ArgumentMatchers.*;
|
||||
|
||||
class QuanxiManagementTest {
|
||||
@Test void productionConstructorsRegisterBothVendorsWithoutExternalSystems() {
|
||||
try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
|
||||
ApiPullProperties props = new ApiPullProperties(); props.setAllowedHosts("192.0.2.10");
|
||||
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.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"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test void createsPausedTaskAndInitialCheckpointWithRealAdapterValidation() throws Exception {
|
||||
QuanxiApiAdapterTest.Fixture f = new QuanxiApiAdapterTest.Fixture();
|
||||
DeviceCollectTaskMapper tasks = mock(DeviceCollectTaskMapper.class);
|
||||
DeviceCollectApiMapper mapper = mock(DeviceCollectApiMapper.class);
|
||||
DeviceDeviceService devices = mock(DeviceDeviceService.class);
|
||||
DeviceDevice device = new DeviceDevice(); device.setId(7); device.setDeviceCollectId(3);
|
||||
device.setIp("192.0.2.10"); when(devices.getByIdSafely(7)).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.singletonMap("api_token", new SecretValue("synthetic-token".toCharArray())));
|
||||
AtomicReference<DeviceCollectApiConfig> config = new AtomicReference<>();
|
||||
AtomicReference<DeviceCollectApiCheckpoint> checkpoint = new AtomicReference<>();
|
||||
when(tasks.insertApiPullTask(any())).thenAnswer(i -> { ((DeviceCollectTask)i.getArgument(0)).setId(11); return 1; });
|
||||
when(mapper.insertConfig(any())).thenAnswer(i -> { config.set(i.getArgument(0)); return 1; });
|
||||
when(mapper.insertCheckpoint(any())).thenAnswer(i -> { checkpoint.set(i.getArgument(0)); return 1; });
|
||||
when(mapper.selectConfig(11)).thenAnswer(i -> config.get());
|
||||
when(mapper.selectCheckpoint(11)).thenAnswer(i -> checkpoint.get());
|
||||
ApiPullTaskService service = new ApiPullTaskService(tasks, mapper, devices,
|
||||
new AdapterRegistry(Collections.singletonList(f.adapter)), credentials, new SafeBaseUrlValidator(props),
|
||||
mock(ApiPullScheduler.class), app, props, new ObjectMapper());
|
||||
CreateApiPullTaskRequest request = new CreateApiPullTaskRequest();
|
||||
request.setDeviceId(7); request.setTaskName("fixture-quanxi"); request.setAdapterType(QuanxiApiAdapter.TYPE);
|
||||
request.setStreamKey(QuanxiApiAdapter.STREAM); request.setBaseUrl(f.config.getBaseUrl());
|
||||
request.setCredentialRef("fixture"); request.setInitialStartTime(OffsetDateTime.parse("2026-08-10T00:00:00+08:00"));
|
||||
request.setExtraConfig(new ObjectMapper().readValue(f.config.getExtraConfigJson(), Map.class));
|
||||
ApiPullTaskResponse response = service.create(request);
|
||||
assertFalse(response.isEnabled()); assertEquals(11, response.getTaskId());
|
||||
assertEquals(QuanxiApiAdapter.TYPE, checkpoint.get().getAdapterType());
|
||||
assertEquals(request.getInitialStartTime(), response.getCommittedWindowEnd());
|
||||
verifyNoInteractions(f.http);
|
||||
}
|
||||
|
||||
@Test void managementRpcFailureDoesNotClaimAuthenticationSuccess() throws Exception {
|
||||
QuanxiApiAdapterTest.Fixture f = new QuanxiApiAdapterTest.Fixture();
|
||||
f.root.remove("result"); f.root.putObject("error").put("code", -32000);
|
||||
f.config.setConfigVersion(1L); f.config.setAdapterVersion(1);
|
||||
DeviceCollectApiMapper mapper = mock(DeviceCollectApiMapper.class);
|
||||
when(mapper.selectConfig(11)).thenReturn(f.config);
|
||||
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());
|
||||
ValidateApiPullTaskResponse response = service.validate(11, 1L);
|
||||
assertFalse(response.isValid());
|
||||
Map<String, ValidationCheckResponse> checks = new HashMap<>();
|
||||
response.getChecks().forEach(c -> checks.put(c.getName(), c));
|
||||
assertEquals("FAIL", checks.get("MINIMAL_ALERT_QUERY").getStatus());
|
||||
assertEquals("RPC_ERROR", checks.get("MINIMAL_ALERT_QUERY").getErrorCode());
|
||||
assertNotEquals("PASS", checks.get("AUTHENTICATION").getStatus());
|
||||
assertFalse(checks.containsKey("MINIMAL_FLOW_QUERY"));
|
||||
}
|
||||
|
||||
@Test void tokenCreationRotationLoadAndSerializationHonorVendorSchema() throws Exception {
|
||||
DeviceApiCredentialMapper mapper = mock(DeviceApiCredentialMapper.class);
|
||||
DeviceDeviceService devices = mock(DeviceDeviceService.class);
|
||||
DeviceDevice device = new DeviceDevice(); device.setId(7); device.setDeviceCollectId(3);
|
||||
when(devices.getByIdSafely(7)).thenReturn(device);
|
||||
AdapterRegistry registry = mock(AdapterRegistry.class);
|
||||
AppProperties app = new AppProperties(); app.setAppServiceDeviceCollectId(3);
|
||||
AtomicReference<DeviceApiCredential> stored = new AtomicReference<>();
|
||||
when(mapper.insert(any())).thenAnswer(i -> { stored.set(i.getArgument(0)); return 1; });
|
||||
when(mapper.selectByRef("fixture")).thenAnswer(i -> stored.get());
|
||||
when(mapper.updateCas(any())).thenAnswer(i -> {
|
||||
DeviceApiCredential value = i.getArgument(0); value.setDeviceId(7); stored.set(value); return 1;
|
||||
});
|
||||
when(mapper.selectForLoad("fixture", 7, "QUANXI_V1")).thenAnswer(i -> stored.get());
|
||||
ApiCredentialService service = new ApiCredentialService(mapper, devices, registry, app);
|
||||
CreateApiCredentialRequest request = new CreateApiCredentialRequest();
|
||||
request.setDeviceId(7); request.setCredentialRef("fixture"); request.setCredentialVersion("v1");
|
||||
request.setAdapterType("QUANXI_V1"); request.setApiToken("synthetic-unit-secret");
|
||||
ApiCredentialResponse response = service.create(request);
|
||||
assertNull(stored.get().getAppKey()); assertNull(stored.get().getAppSecret());
|
||||
assertFalse(request.toString().contains("synthetic-unit-secret"));
|
||||
assertFalse(new ObjectMapper().writeValueAsString(request).contains("synthetic-unit-secret"));
|
||||
assertFalse(new ObjectMapper().writeValueAsString(response).contains("apiToken"));
|
||||
UpdateApiCredentialRequest update = new UpdateApiCredentialRequest();
|
||||
update.setConfigVersion(1); update.setCredentialVersion("v2"); update.setApiToken("synthetic-rotated");
|
||||
service.update("fixture", update);
|
||||
DatabaseCredentialProvider provider = new DatabaseCredentialProvider(mapper);
|
||||
CredentialSnapshot snapshot = provider.load("fixture", "QUANXI_V1", 7);
|
||||
assertArrayEquals("synthetic-rotated".toCharArray(), snapshot.require("api_token").copy());
|
||||
assertThrows(CredentialException.class, () -> snapshot.require("app_secret"));
|
||||
snapshot.close(); assertEquals('\0', snapshot.require("api_token").copy()[0]);
|
||||
assertThrows(CredentialException.class, () -> provider.load("fixture", "QUANXI_V1", 8));
|
||||
request.setAppKey("wrong-vendor-field");
|
||||
assertThrows(CredentialException.class, () -> service.create(request));
|
||||
update.setApiToken("bad\r\nheader");
|
||||
assertThrows(CredentialException.class, () -> service.update("fixture", update));
|
||||
assertThrows(CredentialException.class, () -> CredentialSchemas.validate("NDRSERVER_V1", null, null, "token"));
|
||||
assertThrows(CredentialException.class, () -> CredentialSchemas.validate("QUANXI_V1", null, null, ""));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user