getActiveDevices() {
return deviceDeviceMapper.selectActiveDevices();
}
-}
\ No newline at end of file
+
+ @Override
+ @Transactional
+ public DeviceDevice updateCollectionMode(Integer id, DeviceCollectionMode collectionMode) {
+ if (id == null || collectionMode == null) {
+ throw new IllegalArgumentException("设备 ID 和采集模式不能为空");
+ }
+ DeviceDevice current = deviceDeviceMapper.selectById(id);
+ if (current == null || current.getDeletedAt() != null
+ || (current.getDelFlag() != null && !"0".equals(current.getDelFlag()))) {
+ return null;
+ }
+ if (deviceDeviceMapper.updateCollectionMode(id, collectionMode) != 1) return null;
+ evictDeviceCacheEntries(id, current.getIp());
+ // 采集模式更新属于策略变更，继续保持全量清空 Spring device Cache 的强一致行为。
+ Cache springDeviceCache = cacheManager.getCache("device");
+ if (springDeviceCache != null) springDeviceCache.clear();
+ return deviceDeviceMapper.selectById(id);
+ }
+
+ @Override
+ public DeviceDevice evictDeviceCaches(Integer id) {
+ if (id == null) throw new IllegalArgumentException("设备 ID 不能为空");
+ // 必须绕过二级缓存读取当前 IP，否则可能根据陈旧缓存删除错误的 IP Key。
+ DeviceDevice device = deviceDeviceMapper.selectById(id);
+ if (device == null) {
+ // 即使数据库已无记录，也删除仍可能残留的 ID Key。
+ safeCacheUtil.evict("device:id:" + id);
+ Cache springDeviceCache = cacheManager.getCache("device");
+ if (springDeviceCache != null) springDeviceCache.evict("device:id:" + id);
+ return null;
+ }
+ evictDeviceCacheEntries(id, device.getIp());
+ return device;
+ }
+
+ @Override
+ public void evictAllDeviceCaches() {
+ safeCacheUtil.evictMatching("device:id:*", "device:ip:*");
+ Cache springDeviceCache = cacheManager.getCache("device");
+ if (springDeviceCache != null) springDeviceCache.clear();
+ }
+
+ private void evictDeviceCacheEntries(Integer id, String ip) {
+ String idKey = "device:id:" + id;
+ if (ip == null || ip.trim().isEmpty()) {
+ safeCacheUtil.evict(idKey);
+ } else {
+ safeCacheUtil.evict(idKey, "device:ip:" + ip);
+ }
+ Cache springDeviceCache = cacheManager.getCache("device");
+ if (springDeviceCache == null) return;
+ springDeviceCache.evict(idKey);
+ if (ip != null && !ip.trim().isEmpty()) springDeviceCache.evict("device:ip:" + ip);
+ }
+}
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java b/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java
new file mode 100644
index 0000000..0f76ac6
--- /dev/null
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/config/SyslogBuildStartupLogger.java
@@ -0,0 +1,46 @@
+package com.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Component;
+
+/** 在应用完全就绪后打印不可由现场配置覆盖的代码修订标识和关键开关。 */
+@Slf4j
+@Component
+public class SyslogBuildStartupLogger {
+
+ // 确认 JAR 包含设备级采集模式、API 拉取及全局 TLS 策略。
+ static final String CODE_REVISION = "DEVICE_COLLECTION_API_TLS_V20260917_63";
+ private final Environment environment;
+
+ public SyslogBuildStartupLogger(Environment environment) {
+ this.environment = environment;
+ }
+
+ @EventListener(ApplicationReadyEvent.class)
+ public void onApplicationReady() {
+ log.info("[SYSLOG_BUILD_READY] {}", startupSummary());
+ }
+
+ String startupSummary() {
+ return "codeRevision=" + CODE_REVISION
+ + ", application=syslog-serve"
+ + ", profiles=" + String.join(",", environment.getActiveProfiles())
+ + ", serverPort=" + value("server.port", "8189")
+ + ", contextPath=" + value("server.servlet.context-path", "/syslogserve")
+ + ", syslogEnabled=" + value("syslog.server.enabled", "true")
+ + ", apiPullEnabled=" + value("device.api.enabled", "true")
+ + ", apiTlsVerifyEnabled=" + value("device.api.tls-verify-enabled", "false")
+ + ", apiHostAllowlistEnabled=" + value("device.api.host-allowlist-enabled", "false")
+ + ", walEnabled=" + value("syslog.wal.enabled", "true")
+ + ", terminalRecordEnabled=" + value("syslog.dispatch.terminal-record-enabled", "false")
+ + ", deviceCollectId=" + value("app.service.device_collect_id", "7")
+ + ", kafkaTopic=" + value("spring.kafka.producer.topic", "test-topic");
+ }
+
+ private String value(String key, String fallback) {
+ return environment.getProperty(key, fallback);
+ }
+}
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullProperties.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullProperties.java
index 2ec8692..452026b 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullProperties.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullProperties.java
@@ -7,13 +7,14 @@ import org.springframework.stereotype.Component;
/**
* 进程级 API 拉取开关和资源上限。
*
- * 这里只保存所有任务共享的运行保护参数；页大小、时间窗口、厂商适配器等任务级参数 固化在达梦数据库中。总开关默认关闭，避免系统升级后未经验证就访问真实设备。
+ *
这里只保存所有任务共享的运行保护参数；页大小、时间窗口、厂商适配器等任务级参数固化在达梦数据库中。
+ * 总开关默认开启，但新建任务保持暂停，必须完成校验并 resume 后才会访问真实设备。
*/
@Data
@Component
public class ApiPullProperties {
- @Value("${device.api.enabled:false}")
- private boolean enabled;
+ @Value("${device.api.enabled:true}")
+ private boolean enabled = true;
@Value("${device.api.scheduler.instance-id:${HOSTNAME:local}-${random.uuid}}")
private String instanceId;
@@ -48,6 +49,19 @@ public class ApiPullProperties {
@Value("${device.api.allowed-hosts:}")
private String allowedHosts;
+ /** 现场设备经常增减；默认关闭静态主机白名单，启用后才检查 allowedHosts。 */
+ @Value("${device.api.host-allowlist-enabled:false}")
+ private boolean hostAllowlistEnabled = false;
+
+ /**
+ * 所有设备 API HTTPS 是否校验证书链和主机名。
+ *
+ *
现场默认为内网自签名环境，因此默认关闭。该开关是进程级全局策略：关闭后微步、NDR、全悉等
+ * 所有 API 拉取适配器都不校验证书链和主机名；HTTPS 协议限制和禁止重定向仍然生效。
+ */
+ @Value("${device.api.tls-verify-enabled:false}")
+ private boolean tlsVerifyEnabled = false;
+
/** 在调度器创建线程池前快速失败，防止以危险的并发或租约参数启动。 */
public void validate() {
if (scanIntervalMs < 100
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java
index fbb39ba..13d87b7 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullScheduler.java
@@ -1,6 +1,7 @@
package com.haobang.apipull;
import com.common.mapper.DeviceCollectApiMapper;
+import com.haobang.dispatch.DeviceContextResolver;
import com.haobang.apipull.persistence.DeviceCollectApiCheckpoint;
import com.haobang.apipull.persistence.DeviceCollectApiConfig;
import com.haobang.apipull.persistence.DeviceCollectApiRun;
@@ -31,6 +32,7 @@ public class ApiPullScheduler {
private final DeviceCollectApiMapper mapper;
private final ApiPullRunner runner;
private final SyslogWalService wal;
+ private final DeviceContextResolver deviceResolver;
private ExecutorService executor;
private Semaphore global;
private final ConcurrentHashMap hosts = new ConcurrentHashMap<>();
@@ -40,11 +42,13 @@ public class ApiPullScheduler {
ApiPullProperties properties,
DeviceCollectApiMapper mapper,
ApiPullRunner runner,
- SyslogWalService wal) {
+ SyslogWalService wal,
+ DeviceContextResolver deviceResolver) {
this.properties = properties;
this.mapper = mapper;
this.runner = runner;
this.wal = wal;
+ this.deviceResolver = deviceResolver;
}
@PostConstruct
@@ -110,6 +114,7 @@ public class ApiPullScheduler {
private DeviceCollectApiRun trySubmit(DeviceCollectApiConfig c, String trigger) {
if (!c.isEnabled()) return null;
+ if (!requireApiCollectionMode(c, "MANUAL".equals(trigger))) return null;
if (wal.getFatalFailure() != null || wal.isDiskCritical() || wal.shouldPauseTcp()) {
mapper.scheduleNext(c.getCollectTaskId(), c.getConfigVersion(), c.getPollIntervalSeconds());
return null;
@@ -167,7 +172,18 @@ public class ApiPullScheduler {
executor.execute(
() -> {
try {
- runner.execute(c, leasedCheckpoint, run);
+ // 模式可能在排队期间变化；执行 HTTP 前再次校验，避免已提交线程绕过新策略。
+ if (requireApiCollectionMode(c, false)) {
+ runner.execute(c, leasedCheckpoint, run);
+ } else {
+ run.setStatus("FAILED");
+ run.setErrorCode("COLLECTION_MODE_DENIED");
+ run.setErrorSummary("设备采集模式不允许 API_PULL");
+ run.setFinishedAt(LocalDateTime.now(ApiPullTime.DATABASE_ZONE));
+ mapper.updateRun(run);
+ mapper.releaseLease(
+ c.getCollectTaskId(), properties.getInstanceId(), leasedCheckpoint.getFencingToken());
+ }
} finally {
releaseLocal(c.getCollectTaskId(), hostPermit);
}
@@ -188,6 +204,24 @@ public class ApiPullScheduler {
}
}
+ /**
+ * 调度扫描遇到禁止 API 的设备时推进 next_run_at，避免同一到期任务按 5 秒扫描周期空转。
+ */
+ private boolean requireApiCollectionMode(DeviceCollectApiConfig c, boolean failRequest) {
+ DeviceContextResolver.Resolution resolution = deviceResolver.resolveDeviceId(c.getDeviceId());
+ if (resolution.isRegistered()) return true;
+ mapper.scheduleNext(c.getCollectTaskId(), c.getConfigVersion(), c.getPollIntervalSeconds());
+ String reason = resolution.getReason() == null ? "设备不允许 API_PULL" : resolution.getReason();
+ log.warn("设备 API 拉取已被设备策略跳过: taskId={}, deviceId={}, type={}, reason={}",
+ c.getCollectTaskId(), c.getDeviceId(), resolution.getType(), reason);
+ if (failRequest) {
+ String code = resolution.getType() == DeviceContextResolver.Type.COLLECTION_MODE_DENIED
+ ? "COLLECTION_MODE_DENIED" : "DEVICE_CONTEXT_DENIED";
+ throw new ApiPullException(code, reason, org.springframework.http.HttpStatus.CONFLICT);
+ }
+ return false;
+ }
+
private void releaseLocal(Integer taskId, Semaphore host) {
if (host != null) host.release();
global.release();
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 f989ce3..483ad13 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
@@ -1,6 +1,7 @@
package com.haobang.apipull;
import com.common.entity.DeviceCollectTask;
+import com.common.entity.DeviceCollectionMode;
import com.common.entity.DeviceDevice;
import com.common.mapper.DeviceCollectApiMapper;
import com.common.mapper.DeviceCollectTaskMapper;
@@ -81,11 +82,7 @@ public class ApiPullTaskService {
@Transactional
public ApiPullTaskResponse create(CreateApiPullTaskRequest request) {
// 设备必须归属当前探针，避免一个采集实例替其他探针访问设备内网地址。
- DeviceDevice device = devices.getByIdSafely(request.getDeviceId());
- if (device == null) throw error("INVALID_REQUEST", "设备不存在", HttpStatus.BAD_REQUEST);
- if (device.getDeviceCollectId() == null
- || device.getDeviceCollectId() != app.getAppServiceDeviceCollectId())
- throw error("FORBIDDEN", "设备不属于当前采集探针", HttpStatus.FORBIDDEN);
+ DeviceDevice device = requireApiAllowedDevice(request.getDeviceId());
DeviceCollectApiConfig config = fromCreate(request);
ApiPullAdapter adapter = adapters.require(config.getAdapterType());
adapter.validateConfig(config);
@@ -237,6 +234,7 @@ public class ApiPullTaskService {
DeviceCollectApiConfig c = requireConfig(taskId);
if (c.getConfigVersion() != version)
throw error("CONFIG_VERSION_CONFLICT", "配置版本冲突", HttpStatus.CONFLICT);
+ requireApiAllowedDevice(c.getDeviceId());
return scheduler.trigger(taskId);
}
@@ -245,6 +243,8 @@ public class ApiPullTaskService {
DeviceCollectApiConfig c = requireConfig(taskId);
if (c.getConfigVersion() != version)
throw error("CONFIG_VERSION_CONFLICT", "配置版本冲突", HttpStatus.CONFLICT);
+ // 必须在创建验证审计和发出任何 HTTP 请求前执行设备级策略。
+ requireApiAllowedDevice(c.getDeviceId());
DeviceCollectApiRun audit = new DeviceCollectApiRun();
audit.setRunId(UUID.randomUUID().toString());
audit.setCollectTaskId(taskId);
@@ -301,7 +301,7 @@ public class ApiPullTaskService {
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("TLS_HANDSHAKE", "PASS", null, tlsValidationSummary(), pullStarted));
checks.add(check("AUTHENTICATION", "PASS", null, "厂商认证及查询通过", pullStarted));
checks.add(check("API_PERMISSION", "PASS", null, "业务接口可访问", pullStarted));
checks.add(
@@ -323,7 +323,7 @@ public class ApiPullTaskService {
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("TLS_HANDSHAKE", "PASS", null, tlsValidationSummary(), 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));
@@ -362,6 +362,12 @@ public class ApiPullTaskService {
return r;
}
+ private String tlsValidationSummary() {
+ return apiProperties.isTlsVerifyEnabled()
+ ? "TLS 证书链和主机名校验通过"
+ : "TLS 连接通过（全局已关闭证书链和主机名校验）";
+ }
+
public ApiPullRuntimeResponse runtime(Integer taskId) {
DeviceCollectApiConfig c = requireConfig(taskId);
DeviceCollectApiCheckpoint cp = mapper.selectCheckpoint(taskId);
@@ -529,6 +535,26 @@ public class ApiPullTaskService {
return c;
}
+ private DeviceDevice requireApiAllowedDevice(Integer deviceId) {
+ DeviceDevice device = devices.getByIdSafely(deviceId);
+ if (device == null || device.getDeletedAt() != null
+ || (device.getDelFlag() != null && !"0".equals(device.getDelFlag()))) {
+ throw error("INVALID_REQUEST", "设备不存在或已删除", HttpStatus.BAD_REQUEST);
+ }
+ if (device.getDeviceCollectId() == null
+ || device.getDeviceCollectId() != app.getAppServiceDeviceCollectId()) {
+ throw error("FORBIDDEN", "设备不属于当前采集探针", HttpStatus.FORBIDDEN);
+ }
+ DeviceCollectionMode mode = device.getCollectionMode();
+ if (!mode.allowsApi()) {
+ throw error(
+ "COLLECTION_MODE_DENIED",
+ "设备采集模式不允许 API_PULL: deviceId=" + deviceId + ", collectionMode=" + mode,
+ HttpStatus.CONFLICT);
+ }
+ return device;
+ }
+
private DeviceCollectTask requireTask(Integer id) {
DeviceCollectTask t = tasks.selectById(id);
if (t == null || t.getMethod() == null || t.getMethod() != API_PULL_METHOD)
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 6fb1bc0..fe238e9 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
@@ -3,6 +3,7 @@ package com.haobang.apipull.adapter.ndr;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.haobang.apipull.ApiPullProperties;
import com.haobang.apipull.adapter.AdapterException;
import com.haobang.apipull.adapter.ApiPullAdapter;
import com.haobang.apipull.adapter.PullPage;
@@ -42,8 +43,12 @@ import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.ssl.SSLContextBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -104,8 +109,11 @@ public class NdrServerApiAdapter implements ApiPullAdapter {
*/
@Autowired
public NdrServerApiAdapter(
- CredentialProvider credentials, SafeBaseUrlValidator urlValidator, ObjectMapper mapper) {
- this(credentials, urlValidator, mapper, HttpClients.custom().disableAutomaticRetries().build());
+ CredentialProvider credentials,
+ SafeBaseUrlValidator urlValidator,
+ ObjectMapper mapper,
+ ApiPullProperties properties) {
+ this(credentials, urlValidator, mapper, buildHttpClient(properties.isTlsVerifyEnabled()));
}
NdrServerApiAdapter(
@@ -119,6 +127,24 @@ public class NdrServerApiAdapter implements ApiPullAdapter {
this.http = http;
}
+ /**
+ * 遵循设备 API 全局 TLS 校验开关，但始终禁止自动重试和重定向。
+ */
+ static CloseableHttpClient buildHttpClient(boolean tlsVerifyEnabled) {
+ HttpClientBuilder builder =
+ HttpClients.custom().disableAutomaticRetries().disableRedirectHandling();
+ if (tlsVerifyEnabled) return builder.build();
+ try {
+ javax.net.ssl.SSLContext sslContext =
+ SSLContextBuilder.create().loadTrustMaterial((chain, authType) -> true).build();
+ SSLConnectionSocketFactory socketFactory =
+ new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE);
+ return builder.setSSLSocketFactory(socketFactory).build();
+ } catch (Exception e) {
+ throw new IllegalStateException("初始化 NDR 内网 TLS 客户端失败", e);
+ }
+ }
+
@Override
public String type() {
return TYPE;
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
index 1925bc0..ea20173 100644
--- 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
@@ -5,6 +5,7 @@ import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.haobang.apipull.ApiPullProperties;
import com.haobang.apipull.adapter.*;
import com.haobang.apipull.credential.*;
import com.haobang.apipull.persistence.DeviceCollectApiConfig;
@@ -26,8 +27,12 @@ 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.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.ssl.SSLContextBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -49,9 +54,12 @@ public class QuanxiApiAdapter implements ApiPullAdapter {
private final CloseableHttpClient http;
@Autowired
- public QuanxiApiAdapter(CredentialProvider credentials, SafeBaseUrlValidator urls, ObjectMapper json) {
- this(credentials, urls, json,
- HttpClients.custom().disableAutomaticRetries().disableRedirectHandling().build());
+ public QuanxiApiAdapter(
+ CredentialProvider credentials,
+ SafeBaseUrlValidator urls,
+ ObjectMapper json,
+ ApiPullProperties properties) {
+ this(credentials, urls, json, buildHttpClient(properties.isTlsVerifyEnabled()));
}
QuanxiApiAdapter(CredentialProvider credentials, SafeBaseUrlValidator urls,
@@ -63,6 +71,22 @@ public class QuanxiApiAdapter implements ApiPullAdapter {
this.http = http;
}
+ /** 遵循设备 API 全局 TLS 校验开关，但始终禁止自动重试和重定向。 */
+ static CloseableHttpClient buildHttpClient(boolean tlsVerifyEnabled) {
+ HttpClientBuilder builder =
+ HttpClients.custom().disableAutomaticRetries().disableRedirectHandling();
+ if (tlsVerifyEnabled) return builder.build();
+ try {
+ javax.net.ssl.SSLContext sslContext =
+ SSLContextBuilder.create().loadTrustMaterial((chain, authType) -> true).build();
+ SSLConnectionSocketFactory socketFactory =
+ new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE);
+ return builder.setSSLSocketFactory(socketFactory).build();
+ } catch (Exception e) {
+ throw new IllegalStateException("初始化全悉内网 TLS 客户端失败", e);
+ }
+ }
+
@Override public String type() { return TYPE; }
@Override public int version() { return 1; }
@Override public Set streams() { return Collections.singleton(STREAM); }
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java
new file mode 100644
index 0000000..45120fd
--- /dev/null
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/adapter/weibu/WeibuApiAdapter.java
@@ -0,0 +1,481 @@
+package com.haobang.apipull.adapter.weibu;
+
+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.ApiPullProperties;
+import com.haobang.apipull.adapter.AdapterException;
+import com.haobang.apipull.adapter.ApiPullAdapter;
+import com.haobang.apipull.adapter.PullPage;
+import com.haobang.apipull.adapter.PullRequest;
+import com.haobang.apipull.adapter.PulledEvent;
+import com.haobang.apipull.credential.CredentialException;
+import com.haobang.apipull.credential.CredentialProvider;
+import com.haobang.apipull.credential.CredentialSchemas;
+import com.haobang.apipull.credential.CredentialSnapshot;
+import com.haobang.apipull.persistence.DeviceCollectApiConfig;
+import com.haobang.apipull.security.SafeBaseUrlValidator;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+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.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.PreDestroy;
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+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.client.utils.URIBuilder;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * 微步 API 配置文档中的逐条日志查询协议。
+ *
+ * 固定使用 {@code /api/v1/log/searchBySql}，响应 {@code data.data[]} 中每个对象作为一条原始
+ * 日志进入 WAL/Kafka。该接口没有 offset/page 参数，因此任务强制 {@code maxPagesPerRun=1}；通用
+ * Runner 先读取响应 {@code total}，当窗口内总数超过 {@code pageSize} 时只丢弃探测响应并继续缩小
+ * 时间窗口，直到单次响应能够完整返回。最小一秒窗口仍超限时失败且不推进 checkpoint。
+ *
+ *
externalEventId 使用厂商日志 {@code id}，事件时间使用 {@code time}。事件 ID 只由任务命名空间
+ * 和厂商稳定 ID 生成，lookback 重叠查询不会制造新事件版本。适配器不把 API JSON 伪装成 Syslog，
+ * 业务字段标准化由下游针对 API 的泛化规则完成。
+ */
+@Component
+public class WeibuApiAdapter implements ApiPullAdapter {
+ public static final String TYPE = "WEIBU_V1";
+ public static final String LOG = "WEIBU_LOG";
+ static final String PATH = "/api/v1/log/searchBySql";
+ private static final ZoneId DEVICE_ZONE = ZoneId.of("Asia/Shanghai");
+ private static final DateTimeFormatter EVENT_ID_DATE =
+ DateTimeFormatter.ofPattern("yyyyMMdd").withZone(DEVICE_ZONE);
+ private static final UUID EVENT_NAMESPACE =
+ UUID.fromString("89a53d59-bb84-5d41-924f-38319448cf46");
+ private static final long MIN_EVENT_EPOCH_SECOND = 946684800L;
+ private static final long MAX_EVENT_EPOCH_SECOND = 4102444800L;
+
+ private final CredentialProvider credentials;
+ private final SafeBaseUrlValidator urls;
+ private final ObjectMapper json;
+ private final CloseableHttpClient http;
+
+ @Autowired
+ public WeibuApiAdapter(
+ CredentialProvider credentials,
+ SafeBaseUrlValidator urls,
+ ObjectMapper json,
+ ApiPullProperties properties) {
+ this(
+ credentials,
+ urls,
+ json,
+ buildHttpClient(properties.isTlsVerifyEnabled()));
+ }
+
+ WeibuApiAdapter(
+ 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;
+ }
+
+ /**
+ * 进程级全局开关可为使用自签名证书的内网设备关闭证书链和主机名校验。
+ * 客户端仍拒绝重定向和非 HTTPS URL；静态主机白名单是否启用由部署配置决定。
+ */
+ static CloseableHttpClient buildHttpClient(boolean tlsVerifyEnabled) {
+ HttpClientBuilder builder =
+ HttpClients.custom().disableAutomaticRetries().disableRedirectHandling();
+ if (tlsVerifyEnabled) return builder.build();
+ try {
+ javax.net.ssl.SSLContext sslContext =
+ SSLContextBuilder.create().loadTrustMaterial((chain, authType) -> true).build();
+ SSLConnectionSocketFactory socketFactory =
+ new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE);
+ return builder.setSSLSocketFactory(socketFactory).build();
+ } catch (Exception e) {
+ throw new IllegalStateException("初始化微步内网 TLS 客户端失败", e);
+ }
+ }
+
+ @Override
+ public String type() {
+ return TYPE;
+ }
+
+ @Override
+ public int version() {
+ return 1;
+ }
+
+ @Override
+ public Set streams() {
+ return Collections.singleton(LOG);
+ }
+
+ /** 厂商时间字段为 Unix 秒，窗口不能安全拆到亚秒。 */
+ @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())) fail("CONFIG_SCHEMA_ERROR");
+ if (!LOG.equals(c.getStreamKey())) fail("STREAM_NOT_SUPPORTED");
+ range(c.getPageSize(), 1, 1000);
+ if (c.getMaxPagesPerRun() == null || c.getMaxPagesPerRun() != 1)
+ fail("CONFIG_SCHEMA_ERROR");
+ range(c.getPollIntervalSeconds(), 1, 86400);
+ range(c.getWindowSizeSeconds(), 1, 86400);
+ 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) fail("CONFIG_SCHEMA_ERROR");
+ URI base = urls.validate(c.getBaseUrl());
+ String path = base.getPath();
+ if (!(path == null || path.isEmpty() || "/".equals(path))) fail("CONFIG_SCHEMA_ERROR");
+ if (c.getExtraConfigJson() == null || !"{}".equals(c.getExtraConfigJson().trim()))
+ fail("CONFIG_SCHEMA_ERROR");
+ }
+
+ @Override
+ public PullPage pull(PullRequest request) {
+ DeviceCollectApiConfig config = request.getConfig();
+ validateConfig(config);
+ validateRequest(request);
+
+ Instant vendorStart = request.getQueryStart().truncatedTo(ChronoUnit.SECONDS);
+ Instant vendorEnd = ceilToSecond(request.getQueryEnd()).minusSeconds(1L);
+ if (vendorEnd.isBefore(vendorStart)) fail("REQUEST_RANGE_INVALID");
+
+ ObjectNode body = json.createObjectNode();
+ body.putArray("assets_group");
+ body.put("search_connection", false);
+ body.putArray("net_data_type").add("attack").add("risk");
+ body.put("time_from", vendorStart.getEpochSecond());
+ body.put("time_to", vendorEnd.getEpochSecond());
+ body.put("sql", "");
+ body.putArray("columns");
+ body.put("asc", false);
+ // 厂商接口只有 size，没有 offset/page。Runner 通过 total 自动缩小窗口，确保该次
+ // size 足以取回窗口内全部日志，而不是把 size 当作伪分页游标。
+ body.put("size", request.getLimit());
+
+ byte[] responseBytes = null;
+ HttpPost post = null;
+ long httpStarted = System.nanoTime();
+ try (CredentialSnapshot snapshot =
+ credentials.load(config.getCredentialRef(), TYPE, config.getDeviceId())) {
+ char[] keyChars = snapshot.require("app_key").copy();
+ char[] secretChars = snapshot.require("app_secret").copy();
+ try {
+ String apiKey = new String(keyChars);
+ String secret = new String(secretChars);
+ CredentialSchemas.validate(TYPE, apiKey, secret, null);
+ long timestamp = Instant.now().getEpochSecond();
+ String signature = sign(apiKey, secretChars, timestamp);
+ URI endpoint =
+ new URIBuilder(endpoint(urls.validate(config.getBaseUrl())))
+ .addParameter("api_key", apiKey)
+ .addParameter("auth_timestamp", Long.toString(timestamp))
+ .addParameter("sign", signature)
+ .build();
+ post = new HttpPost(endpoint);
+ post.setConfig(requestConfig(config));
+ post.setHeader("Accept", "application/json");
+ post.setEntity(
+ new StringEntity(json.writeValueAsString(body), ContentType.APPLICATION_JSON));
+ try (CloseableHttpResponse response = http.execute(post)) {
+ int status = response.getStatusLine().getStatusCode();
+ if (status < 200 || status >= 300) {
+ String code =
+ status == 401
+ ? "AUTH_HTTP_401"
+ : status == 403 ? "AUTH_HTTP_403" : "HTTP_ERROR";
+ throw new AdapterException(code, "微步 API HTTP 请求失败: " + status, status);
+ }
+ responseBytes = readBounded(response.getEntity(), config.getMaxResponseBytes());
+ long httpMs = elapsed(httpStarted);
+ long parseStarted = System.nanoTime();
+ return parsePage(responseBytes, status, httpMs, parseStarted, request);
+ }
+ } finally {
+ Arrays.fill(keyChars, '\0');
+ Arrays.fill(secretChars, '\0');
+ }
+ } catch (AdapterException | CredentialException e) {
+ throw e;
+ } catch (SocketTimeoutException | org.apache.http.conn.ConnectTimeoutException e) {
+ fail("HTTP_TIMEOUT");
+ return null;
+ } catch (IOException e) {
+ fail("DEVICE_API_UNAVAILABLE");
+ return null;
+ } catch (Exception e) {
+ fail("REQUEST_BUILD_ERROR");
+ return null;
+ } finally {
+ if (post != null) post.releaseConnection();
+ if (responseBytes != null) Arrays.fill(responseBytes, (byte) 0);
+ }
+ }
+
+ private PullPage parsePage(
+ byte[] bytes,
+ int status,
+ long httpMs,
+ long parseStarted,
+ PullRequest request) {
+ try {
+ JsonNode root = parseBounded(bytes);
+ if (!root.isObject()) fail("RESPONSE_SCHEMA_ERROR");
+ JsonNode responseCode = root.get("response_code");
+ if (responseCode == null || !responseCode.isIntegralNumber())
+ fail("RESPONSE_SCHEMA_ERROR");
+ if (responseCode.intValue() != 0) fail("VENDOR_RESPONSE_ERROR");
+ JsonNode data = root.get("data");
+ if (data == null || !data.isObject()) fail("RESPONSE_SCHEMA_ERROR");
+ JsonNode items = data.get("data");
+ if (items == null || !items.isArray())
+ fail("RESPONSE_SCHEMA_ERROR");
+ long total = nonNegativeLong(data.get("total"));
+ long expectedItems = Math.min(total, request.getLimit());
+ if (items.size() != expectedItems) fail("PAGINATION_INCONSISTENT");
+
+ ArrayList events = new ArrayList<>();
+ Set pageIds = new HashSet<>();
+ UUID taskNamespace =
+ uuidV5(
+ EVENT_NAMESPACE,
+ configIdentity(request.getConfig()));
+ for (JsonNode item : items) {
+ if (!item.isObject()) fail("RESPONSE_SCHEMA_ERROR");
+ String external = requiredText(item.get("id"), 512);
+ if (!pageIds.add(external)) fail("PAGINATION_DUPLICATE_DOCUMENT");
+ long eventSecond = integral(item.get("time"));
+ if (eventSecond < MIN_EVENT_EPOCH_SECOND || eventSecond >= MAX_EVENT_EPOCH_SECOND)
+ fail("EVENT_TIME_INVALID");
+ Instant eventTime = Instant.ofEpochSecond(eventSecond);
+ // 请求使用外扩后的整秒闭区间，最终仍严格服从平台的半开窗口。厂商若返回
+ // 外扩边界记录，只按设备返回数完成完整性校验，不让该记录进入 WAL。
+ if (eventTime.isBefore(request.getQueryStart())
+ || !eventTime.isBefore(request.getQueryEnd())) continue;
+ String raw = json.writeValueAsString(item);
+ if (raw.getBytes(StandardCharsets.UTF_8).length > 1048576) fail("EVENT_TOO_LARGE");
+ String eventId =
+ EVENT_ID_DATE.format(eventTime) + "-" + uuidV5(taskNamespace, external);
+ events.add(new PulledEvent(eventId, external, eventTime, raw));
+ }
+ return new PullPage(
+ events,
+ total,
+ status,
+ bytes.length,
+ httpMs,
+ elapsed(parseStarted),
+ Long.toString(total),
+ items.size());
+ } catch (AdapterException e) {
+ throw e;
+ } catch (Exception e) {
+ fail("RESPONSE_SCHEMA_ERROR");
+ return null;
+ }
+ }
+
+ private void validateRequest(PullRequest r) {
+ if (r.getConfig().getDeviceId() == null
+ || r.getConfig().getCollectTaskId() == null
+ || r.getQueryStart() == null
+ || r.getQueryEnd() == null
+ || !r.getQueryEnd().isAfter(r.getQueryStart())
+ || r.getLimit() < 1
+ || r.getLimit() > r.getConfig().getPageSize()
+ || r.getOffset() != 0L) fail("REQUEST_RANGE_INVALID");
+ }
+
+ private URI endpoint(URI base) throws Exception {
+ return new URI(base.getScheme(), null, base.getHost(), base.getPort(), PATH, null, null);
+ }
+
+ private RequestConfig requestConfig(DeviceCollectApiConfig c) {
+ return RequestConfig.custom()
+ .setConnectTimeout(c.getConnectTimeoutMs())
+ .setConnectionRequestTimeout(c.getConnectTimeoutMs())
+ .setSocketTimeout(c.getRequestTimeoutMs())
+ .setRedirectsEnabled(false)
+ .build();
+ }
+
+ static String sign(String apiKey, char[] secret, long timestamp) throws Exception {
+ byte[] secretBytes = new String(secret).getBytes(StandardCharsets.UTF_8);
+ try {
+ Mac mac = Mac.getInstance("HmacSHA256");
+ mac.init(new SecretKeySpec(secretBytes, "HmacSHA256"));
+ byte[] digest =
+ mac.doFinal((apiKey + timestamp).getBytes(StandardCharsets.UTF_8));
+ try {
+ return Base64.getUrlEncoder().withoutPadding().encodeToString(digest);
+ } finally {
+ Arrays.fill(digest, (byte) 0);
+ }
+ } finally {
+ Arrays.fill(secretBytes, (byte) 0);
+ }
+ }
+
+ static Instant ceilToSecond(Instant value) {
+ Instant floor = value.truncatedTo(ChronoUnit.SECONDS);
+ return floor.equals(value) ? floor : floor.plusSeconds(1L);
+ }
+
+ private byte[] readBounded(HttpEntity entity, long maxBytes) throws IOException {
+ if (entity == null) fail("RESPONSE_SCHEMA_ERROR");
+ if (entity.getContentLength() > maxBytes) fail("RESPONSE_TOO_LARGE");
+ try (InputStream in = entity.getContent();
+ ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+ byte[] buffer = new byte[8192];
+ long size = 0;
+ int n;
+ while ((n = in.read(buffer)) != -1) {
+ size += n;
+ if (size > maxBytes) fail("RESPONSE_TOO_LARGE");
+ out.write(buffer, 0, n);
+ }
+ return out.toByteArray();
+ }
+ }
+
+ private JsonNode parseBounded(byte[] bytes) {
+ try {
+ try (JsonParser parser = json.getFactory().createParser(bytes)) {
+ int depth = 0;
+ int tokens = 0;
+ JsonToken token;
+ while ((token = parser.nextToken()) != null) {
+ if (++tokens > 2000000) fail("RESPONSE_SCHEMA_ERROR");
+ if (token == JsonToken.START_OBJECT || token == JsonToken.START_ARRAY) {
+ if (++depth > 64) fail("RESPONSE_SCHEMA_ERROR");
+ } else if (token == JsonToken.END_OBJECT || token == JsonToken.END_ARRAY) {
+ depth--;
+ } else if ((token == JsonToken.VALUE_STRING && parser.getTextLength() > 1048576)
+ || (token == JsonToken.FIELD_NAME && parser.getCurrentName().length() > 512)) {
+ fail("RESPONSE_SCHEMA_ERROR");
+ }
+ }
+ }
+ try (JsonParser parser = json.getFactory().createParser(bytes)) {
+ JsonNode value = json.readTree(parser);
+ if (value == null || parser.nextToken() != null) fail("RESPONSE_SCHEMA_ERROR");
+ return value;
+ }
+ } catch (AdapterException e) {
+ throw e;
+ } catch (Exception e) {
+ fail("RESPONSE_SCHEMA_ERROR");
+ return null;
+ }
+ }
+
+ private String configIdentity(DeviceCollectApiConfig c) {
+ return c.getDeviceId() + "|" + c.getCollectTaskId() + "|" + TYPE + "|" + LOG;
+ }
+
+ private long nonNegativeLong(JsonNode value) {
+ long result = integral(value);
+ if (result < 0) fail("RESPONSE_SCHEMA_ERROR");
+ return result;
+ }
+
+ private long integral(JsonNode value) {
+ if (value == null || !value.isIntegralNumber() || !value.canConvertToLong())
+ fail("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) fail("RESPONSE_SCHEMA_ERROR");
+ return value.textValue();
+ }
+
+ private void range(Integer value, int min, int max) {
+ if (value == null || value < min || value > max) fail("CONFIG_SCHEMA_ERROR");
+ }
+
+ static UUID uuidV5(UUID namespace, String value) throws Exception {
+ MessageDigest digest = MessageDigest.getInstance("SHA-1");
+ digest.update(
+ ByteBuffer.allocate(16)
+ .putLong(namespace.getMostSignificantBits())
+ .putLong(namespace.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 buffer = ByteBuffer.wrap(hash);
+ return new UUID(buffer.getLong(), buffer.getLong());
+ }
+
+ private long elapsed(long started) {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ }
+
+ private void fail(String code) {
+ throw new AdapterException(code, "微步 API: " + code);
+ }
+
+ @PreDestroy
+ public void close() {
+ try {
+ http.close();
+ } catch (IOException ignored) {
+ // Spring shutdown best effort.
+ }
+ }
+}
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
index 1fa8448..6f7c992 100644
--- 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
@@ -5,7 +5,7 @@ public final class CredentialSchemas {
private CredentialSchemas() {}
public static void validate(String type, String key, String secret, String token) {
- if ("NDRSERVER_V1".equals(type)) {
+ if ("NDRSERVER_V1".equals(type) || "WEIBU_V1".equals(type)) {
require(key, 512);
require(secret, 2048);
if (token != null) fail();
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/security/SafeBaseUrlValidator.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/security/SafeBaseUrlValidator.java
index 4c0ebfe..f6f6547 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/security/SafeBaseUrlValidator.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/security/SafeBaseUrlValidator.java
@@ -14,14 +14,16 @@ import org.springframework.stereotype.Component;
/**
* 设备基础地址的 SSRF 防护边界。
*
- * 只有显式加入白名单的 HTTPS 主机才能访问；拒绝 user-info、query、fragment 和路径穿越。 主机采用精确匹配，不允许后缀或通配符匹配，避免相似域名绕过。TLS
- * 证书链和主机名继续由 JVM 默认校验器验证。
+ *
始终拒绝非 HTTPS、user-info、query、fragment 和路径穿越。现场默认关闭静态主机白名单，
+ * 避免设备增减必须重启；显式启用后采用精确主机匹配，不允许后缀或通配符匹配。
*/
@Component
public class SafeBaseUrlValidator {
+ private final boolean hostAllowlistEnabled;
private final Set allowedHosts;
public SafeBaseUrlValidator(ApiPullProperties properties) {
+ hostAllowlistEnabled = properties.isHostAllowlistEnabled();
String configured = properties.getAllowedHosts();
allowedHosts =
configured == null || configured.trim().isEmpty()
@@ -45,7 +47,7 @@ public class SafeBaseUrlValidator {
throw new AdapterException(
"SSRF_REJECTED", "baseUrl 必须是无 user-info/query/fragment 的 HTTPS URL");
String host = uri.getHost().toLowerCase();
- if (allowedHosts.isEmpty() || !allowedHosts.contains(host))
+ if (hostAllowlistEnabled && (allowedHosts.isEmpty() || !allowedHosts.contains(host)))
throw new AdapterException("SSRF_REJECTED", "设备主机不在 device.api.allowed-hosts 中: " + host);
InetAddress[] addresses = InetAddress.getAllByName(host);
if (addresses.length == 0)
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java
index 664c704..f79247a 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java
@@ -1,6 +1,7 @@
package com.haobang.dispatch;
import com.common.entity.DeviceDevice;
+import com.common.entity.DeviceCollectionMode;
import com.common.entity.DeviceUnknown;
import com.common.service.DeviceDeviceService;
import com.common.service.DeviceUnknownService;
@@ -80,7 +81,7 @@ public class DeviceContextResolver {
for (Map.Entry> entry : grouped.entrySet()) {
List sameIp = entry.getValue();
Resolution resolution = sameIp.size() == 1
- ? validateAssignment(sameIp.get(0))
+ ? validateSyslogDevice(sameIp.get(0))
: Resolution.terminal(Type.DUPLICATE_DEVICE, null,
"来源 IP 存在多条设备记录: " + entry.getKey());
resolutions.put(entry.getKey(), resolution);
@@ -147,15 +148,7 @@ public class DeviceContextResolver {
resolution = Resolution.terminal(Type.DUPLICATE_DEVICE, null,
"来源 IP 存在多条设备记录: " + sourceIp);
} else {
- DeviceDevice device = devices.get(0);
- Integer assignedCollector = device.getDeviceCollectId();
- if (assignedCollector == null
- || assignedCollector != properties.getAppServiceDeviceCollectId()) {
- resolution = Resolution.terminal(Type.NOT_ASSIGNED, device,
- "设备不属于当前采集探针: deviceId=" + device.getId());
- } else {
- resolution = Resolution.registered(device);
- }
+ resolution = validateSyslogDevice(devices.get(0));
}
cache.put(sourceIp, new CacheEntry(resolution,
now + properties.getSyslogDeviceContextCacheTtlMs()));
@@ -196,9 +189,10 @@ public class DeviceContextResolver {
}
misses.increment();
DeviceDevice device = deviceService.getByIdSafely(deviceId);
- Resolution resolution = device == null
+ Resolution resolution = device == null || device.getDeletedAt() != null
+ || (device.getDelFlag() != null && !"0".equals(device.getDelFlag()))
? Resolution.terminal(Type.UNKNOWN_DEVICE, null, "设备不存在: deviceId=" + deviceId)
- : validateAssignment(device);
+ : validateApiDevice(device);
deviceIdCache.put(deviceId, new CacheEntry(resolution,
now + properties.getSyslogDeviceContextCacheTtlMs()));
return resolution;
@@ -214,10 +208,46 @@ public class DeviceContextResolver {
return Resolution.registered(device);
}
+ private Resolution validateSyslogDevice(DeviceDevice device) {
+ Resolution assignment = validateAssignment(device);
+ if (!assignment.isRegistered()) return assignment;
+ DeviceCollectionMode mode = device.getCollectionMode();
+ if (!mode.allowsSyslog()) {
+ return Resolution.terminal(Type.COLLECTION_MODE_DENIED, device,
+ "设备采集模式不允许 SYSLOG_PUSH: deviceId=" + device.getId()
+ + ", collectionMode=" + mode);
+ }
+ return assignment;
+ }
+
+ private Resolution validateApiDevice(DeviceDevice device) {
+ Resolution assignment = validateAssignment(device);
+ if (!assignment.isRegistered()) return assignment;
+ DeviceCollectionMode mode = device.getCollectionMode();
+ if (!mode.allowsApi()) {
+ return Resolution.terminal(Type.COLLECTION_MODE_DENIED, device,
+ "设备采集模式不允许 API_PULL: deviceId=" + device.getId()
+ + ", collectionMode=" + mode);
+ }
+ return assignment;
+ }
+
public void invalidate(String sourceIp) {
if (sourceIp != null) cache.remove(sourceIp);
}
+ /** 清除指定设备的 device_id 和所有关联 source_ip 本地缓存。 */
+ public void invalidateDevice(Integer deviceId, String currentIp) {
+ if (deviceId != null) deviceIdCache.remove(deviceId);
+ cache.forEach((sourceIp, entry) -> {
+ DeviceDevice cachedDevice = entry.resolution.getDevice();
+ if ((currentIp != null && currentIp.equals(sourceIp))
+ || (deviceId != null && cachedDevice != null && deviceId.equals(cachedDevice.getId()))) {
+ cache.remove(sourceIp, entry);
+ }
+ });
+ }
+
public void invalidateAll() { cache.clear(); deviceIdCache.clear(); }
public long getHitCount() { return hits.sum(); }
public long getMissCount() { return misses.sum(); }
@@ -255,7 +285,13 @@ public class DeviceContextResolver {
}
}
- public enum Type { REGISTERED, UNKNOWN_DEVICE, DUPLICATE_DEVICE, NOT_ASSIGNED }
+ public enum Type {
+ REGISTERED,
+ UNKNOWN_DEVICE,
+ DUPLICATE_DEVICE,
+ NOT_ASSIGNED,
+ COLLECTION_MODE_DENIED
+ }
public enum IngressStatus { REGISTERED, INVALID, UNAVAILABLE }
private static final class IngressSnapshot {
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java
index 4c343bd..226b7f6 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java
@@ -2,6 +2,7 @@ package com.haobang.dispatch;
import com.haobang.config.AppProperties;
import com.haobang.diagnostics.SyslogIpMetricsRegistry;
+import com.haobang.wal.IngressType;
import com.haobang.wal.SyslogWalRecord;
import com.haobang.wal.SyslogWalService;
import com.kafka.kafkaProducer;
@@ -30,9 +31,9 @@ import java.util.concurrent.atomic.LongAdder;
/**
* 从 WAL 有界并发异步投递 Kafka，并只按连续 ACK 推进 WAL 游标。
*
- * v2/v3 保持原有按 source_ip 识别设备和单时间参数的生产路径；v4 改为按可信 device_id
- * 识别设备，并分别传递 receive_time/event_time。兼容分支留在 Dispatcher，Kafka 消息外层结构
- * 不因接入方式改变。
+ *
v2/v3 保持原有按 source_ip 识别设备和单时间参数的生产路径；v4 API_PULL 按可信
+ * device_id 识别设备，并分别传递 receive_time/event_time。Dispatcher 按 ingress_type 选择
+ * 设备策略，确保模式变化后已经进入 WAL 的记录仍会被最终校验。
*/
@Component
public class SyslogWalDispatcher {
@@ -148,8 +149,10 @@ public class SyslogWalDispatcher {
}
final DeviceContextResolver.Resolution resolution;
try {
- // API 的 HTTP 对端可能是代理或网关；v4 必须使用任务写入 WAL 的 device_id。
- if (record.getSchemaVersion() >= 4 && record.getDeviceId() != null) {
+ // API 的 HTTP 对端可能是代理或网关；API_PULL 必须使用任务写入 WAL 的 device_id。
+ // 不能仅凭 schema/device_id 判断，否则未来携带 device_id 的 SYSLOG_PUSH 会误用 API 权限。
+ if (record.getSchemaVersion() >= 4
+ && IngressType.API_PULL.name().equals(record.getIngressType())) {
resolution = deviceResolver.resolveDeviceId(record.getDeviceId());
} else {
resolution = deviceResolver.resolve(record.getSourceIp(), record.getProtocol());
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/util/SafeCacheUtil.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/util/SafeCacheUtil.java
index 26eef7b..11e0e97 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/util/SafeCacheUtil.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/util/SafeCacheUtil.java
@@ -9,6 +9,11 @@ import java.util.function.Supplier;
import java.util.LinkedHashMap;
import java.util.List ;
import java.util.stream.Collectors;
+import java.util.Arrays;
+import java.util.ArrayList;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.ScanOptions;
@Component
public class SafeCacheUtil {
@@ -105,4 +110,39 @@ public class SafeCacheUtil {
return loader.get();
}
}
-}
\ No newline at end of file
+
+ /**
+ * 严格删除 SafeCacheUtil 使用的原始 Redis Key。
+ *
+ *
设备策略更新不能像读缓存一样静默降级，否则接口成功后仍可能读取旧策略。
+ */
+ public void evict(String... keys) {
+ if (keys == null || keys.length == 0) return;
+ redisTemplate.delete(Arrays.asList(keys));
+ }
+
+ /** 使用 Redis SCAN 分批删除匹配的原始 Key，避免生产环境执行阻塞式 KEYS。 */
+ public void evictMatching(String... patterns) {
+ if (patterns == null || patterns.length == 0) return;
+ redisTemplate.execute((RedisCallback) connection -> {
+ for (String pattern : patterns) {
+ if (pattern == null || pattern.trim().isEmpty()) continue;
+ ScanOptions options = ScanOptions.scanOptions().match(pattern).count(500).build();
+ try (Cursor cursor = connection.scan(options)) {
+ List batch = new ArrayList<>(500);
+ while (cursor.hasNext()) {
+ batch.add(cursor.next());
+ if (batch.size() == 500) {
+ connection.del(batch.toArray(new byte[batch.size()][]));
+ batch.clear();
+ }
+ }
+ if (!batch.isEmpty()) {
+ connection.del(batch.toArray(new byte[batch.size()][]));
+ }
+ }
+ }
+ return null;
+ });
+ }
+}
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 326bf7a..85b6ef9 100644
--- a/haobang-security-dm/syslog-serve/src/main/resources/application.properties
+++ b/haobang-security-dm/syslog-serve/src/main/resources/application.properties
@@ -7,8 +7,8 @@ server.error.include-message=always
server.error.include-binding-errors=always
# Syslog Server Configuration
-# 当前测试只验证设备 API 拉取，关闭 TCP/UDP Netty 监听；可用环境变量重新开启。
-syslog.server.enabled=${SYSLOG_SERVER_ENABLED:false}
+# 默认同时开启 TCP/UDP Syslog 与设备 API 拉取；设备级 collectionMode 再决定具体接入方式。
+syslog.server.enabled=${SYSLOG_SERVER_ENABLED:true}
syslog.tcp.port=514
syslog.udp.port=514
syslog.tcp.backlog=${SYSLOG_TCP_BACKLOG:4096}
@@ -127,7 +127,7 @@ mybatis.mapper-locations=classpath:mapper/*.xml
mybatis.type-aliases-package=com.common.entity
mybatis.configuration.map-underscore-to-camel-case=true
-# 当前测试环境开启设备 API 拉取调度。
+# 默认开启设备 API 拉取调度；单个任务仍须通过 resume 启用。
device.api.enabled=${DEVICE_API_ENABLED:true}
# 设备 AppKey/AppSecret 由 device_collect_api_credential 表管理，配置文件不再保存厂商凭据。
device.api.scheduler.instance-id=${DEVICE_API_INSTANCE_ID:${HOSTNAME:local}}
@@ -141,9 +141,12 @@ device.api.scheduler.max-pages-per-slice=${DEVICE_API_MAX_PAGES_PER_SLICE:20}
# HTTP + 解析 + 本页 WAL durable 总耗时达到 1 秒即打印一条结构化 WARN。
device.api.page.slow-threshold-ms=${DEVICE_API_SLOW_PAGE_THRESHOLD_MS:1000}
device.api.page.backpressure-timeout-ms=${DEVICE_API_BACKPRESSURE_TIMEOUT_MS:120000}
-# 逗号分隔的设备 DNS 名或 IP；必须显式配置，防止管理接口被用作 SSRF 跳板。
-# NDR 测试设备证书仅声明 CN=wlza.com；通过宿主机/容器 hosts 将该域名固定映射到 192.168.4.19。
+# 现场默认关闭静态主机白名单，新增 API 设备无需修改配置和重启；需要收紧时可显式开启。
+device.api.host-allowlist-enabled=${DEVICE_API_HOST_ALLOWLIST_ENABLED:false}
+# 白名单开启时使用的逗号分隔 DNS 名或 IP；主机精确匹配，不支持通配符。
device.api.allowed-hosts=${DEVICE_API_ALLOWED_HOSTS:192.168.4.19,wlza.com}
+# 所有设备 API 默认关闭证书链和主机名校验；仍只允许 HTTPS，并禁止重定向。
+device.api.tls-verify-enabled=${DEVICE_API_TLS_VERIFY_ENABLED:false}
# 开发环境缓存配置
spring.redis.host=localhost
diff --git a/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceDeviceMapper.xml b/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceDeviceMapper.xml
index 97558fc..f4430b1 100644
--- a/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceDeviceMapper.xml
+++ b/haobang-security-dm/syslog-serve/src/main/resources/mapper/DeviceDeviceMapper.xml
@@ -42,6 +42,9 @@
+
@@ -51,7 +54,8 @@
control_agent_id, license_start_time, license_end_time, is_monitoring,
security_scope_id, owner_id, ssh_config_id, status, created_by_id, decode_type,
miss_policy, tenant_id, create_time, update_time, create_by, update_by, del_flag,
- manager_name, today_parse_count, today_non_log_count, create_dept, device_collect_id
+ manager_name, today_parse_count, today_non_log_count, create_dept, device_collect_id,
+ collection_mode
@@ -235,4 +239,14 @@
WHERE vendor = #{vendor}