feat(syslog): 为三个生产模块增加启动版本标识

- syslog-serve 增加 SYSLOG_BUILD_READY 启动日志
- syslog-consumer 增加 SYSLOG_BUILD_READY 启动日志
- 更新 syslog-consumer-rule 代码修订标识至 v63
- 输出模块名、代码修订号、运行环境及关键非敏感配置
- 支持通过容器日志快速确认实际部署的 JAR 版本
- 补充 syslog-serve 和 syslog-consumer 启动日志测试
- 不输出数据库密码、API 凭据等敏感信息
- 不修改 Maven POM 配置

haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/ApiPullProperties.java

@@ -7,13 +7,14 @@ import org.springframework.stereotype.Component;
 /**
  * 进程级 API 拉取开关和资源上限。
  *
- * <p>这里只保存所有任务共享的运行保护参数；页大小、时间窗口、厂商适配器等任务级参数 固化在达梦数据库中。总开关默认关闭，避免系统升级后未经验证就访问真实设备。
+ * <p>这里只保存所有任务共享的运行保护参数；页大小、时间窗口、厂商适配器等任务级参数固化在达梦数据库中。
+ * 总开关默认开启，但新建任务保持暂停，必须完成校验并 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 是否校验证书链和主机名。
+   *
+   * <p>现场默认为内网自签名环境，因此默认关闭。该开关是进程级全局策略：关闭后微步、NDR、全悉等
+   * 所有 API 拉取适配器都不校验证书链和主机名；HTTPS 协议限制和禁止重定向仍然生效。
+   */
+  @Value("${device.api.tls-verify-enabled:false}")
+  private boolean tlsVerifyEnabled = false;
+
   /** 在调度器创建线程池前快速失败，防止以危险的并发或租约参数启动。 */
   public void validate() {
     if (scanIntervalMs < 100

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<String, Semaphore> 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();

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)

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;

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<String> streams() { return Collections.singleton(STREAM); }

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 配置文档中的逐条日志查询协议。
+ *
+ * <p>固定使用 {@code /api/v1/log/searchBySql}，响应 {@code data.data[]} 中每个对象作为一条原始
+ * 日志进入 WAL/Kafka。该接口没有 offset/page 参数，因此任务强制 {@code maxPagesPerRun=1}；通用
+ * Runner 先读取响应 {@code total}，当窗口内总数超过 {@code pageSize} 时只丢弃探测响应并继续缩小
+ * 时间窗口，直到单次响应能够完整返回。最小一秒窗口仍超限时失败且不推进 checkpoint。
+ *
+ * <p>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<String> 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<PulledEvent> events = new ArrayList<>();
+      Set<String> 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.
+    }
+  }
+}

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();

haobang-security-dm/syslog-serve/src/main/java/com/haobang/apipull/security/SafeBaseUrlValidator.java

@@ -14,14 +14,16 @@ import org.springframework.stereotype.Component;
 /**
  * 设备基础地址的 SSRF 防护边界。
  *
- * <p>只有显式加入白名单的 HTTPS 主机才能访问；拒绝 user-info、query、fragment 和路径穿越。 主机采用精确匹配，不允许后缀或通配符匹配，避免相似域名绕过。TLS
- * 证书链和主机名继续由 JVM 默认校验器验证。
+ * <p>始终拒绝非 HTTPS、user-info、query、fragment 和路径穿越。现场默认关闭静态主机白名单，
+ * 避免设备增减必须重启；显式启用后采用精确主机匹配，不允许后缀或通配符匹配。
  */
 @Component
 public class SafeBaseUrlValidator {
+  private final boolean hostAllowlistEnabled;
   private final Set<String> 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)

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<String, List<DeviceDevice>> entry : grouped.entrySet()) {
                 List<DeviceDevice> 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 {

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 游标。
  *
- * <p>v2/v3 保持原有按 source_ip 识别设备和单时间参数的生产路径；v4 改为按可信 device_id
- * 识别设备，并分别传递 receive_time/event_time。兼容分支留在 Dispatcher，Kafka 消息外层结构
- * 不因接入方式改变。
+ * <p>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());

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();
         }
     }
-}
+
+    /**
+     * 严格删除 SafeCacheUtil 使用的原始 Redis Key。
+     *
+     * <p>设备策略更新不能像读缓存一样静默降级，否则接口成功后仍可能读取旧策略。</p>
+     */
+    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<Void>) connection -> {
+            for (String pattern : patterns) {
+                if (pattern == null || pattern.trim().isEmpty()) continue;
+                ScanOptions options = ScanOptions.scanOptions().match(pattern).count(500).build();
+                try (Cursor<byte[]> cursor = connection.scan(options)) {
+                    List<byte[]> 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;
+        });
+    }
+}