diff --git a/llm_readme.txt b/llm_readme.txt index 6f7a047..a9ebfd5 100644 --- a/llm_readme.txt +++ b/llm_readme.txt @@ -94,6 +94,7 @@ storageNode/ | 6 | `IdentityMessage` | 握手 | | 7 | `MaintainMessage` | 心跳 | | 8 | `AvailableCheckMessage` | 可用性检查 | +| 9 | `SubscriptionSnapshotMessage` | 主站 → 节点，完整订阅备机快照 | --- @@ -107,10 +108,12 @@ storageNode/ ### BackupSubServer（端口 8889） - 提供 V2Ray 和 Clash 代理订阅文件 -- 每 12 小时从配置的 URL 下载最新订阅数据 -- 过滤高比例代理节点 -- 文件存储于 `sub/DouNaiV2ray.txt` 和 `sub/DouNaiClash.txt` -- 通过路径密钥进行身份验证 +- 不直接访问上游；主站完成下载和倍率过滤后，通过 Netty 类型 9 推送完整快照 +- 按公开 Key 的 SHA-256 查找用户绑定的子账号，未知 Key 返回 404 +- 快照经 SHA-256 和 HMAC 校验后原子落盘，主站离线时继续分发最后成功版本 +- 文件存储于 `sub/snapshots/{revision}/accounts/{accountId}/` +- 没有有效快照或快照超过最大有效期时返回 503，不回退旧共享订阅 +- `GET /health/subscription` 提供不含 Key 和订阅正文的快照状态 --- @@ -132,10 +135,18 @@ storageNode/ ### config.properties（从 `/root/gallery/storageNode/config.properties` 加载） ```properties -DouNaiV2ray=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=v2 -DouNaiClash=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=clashmeta +SubscriptionSyncEnabled=false +SubscriptionSyncSecret= +SubscriptionDataDir=/root/gallery/storageNode/sub +SubscriptionMaxStaleSeconds=604800 +SubscriptionMaxPayloadBytes=52428800 +SubscriptionHttpPort=8889 +SubscriptionHttpWorkers=4 +SubscriptionSocketTimeoutMs=10000 ``` +生产同步密钥优先通过 `SUBSCRIPTION_SYNC_SECRET` 环境变量提供，不得提交到仓库或写入日志。 + ### simplelogger.properties - 日志级别：`info` - 时间戳格式：`yyyy-MM-dd HH:mm:ss` @@ -150,4 +161,4 @@ DouNaiClash=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=clashmeta - 外部连接：`lionwebsite.xyz`、`personal.lionwebsite.xyz`、`aaaa.gay` - GraalVM 原生镜像编译，包含 Jackson 反射配置 - 大量使用 Lombok（`@Data`、`@Slf4j`） -- 使用 Hutool 工具库处理文件/ZIP/HTTP 操作 \ No newline at end of file +- 使用 Hutool 工具库处理文件/ZIP/HTTP 操作 diff --git a/src/main/java/lion/Externel/BackupSubServer.java b/src/main/java/lion/Externel/BackupSubServer.java index 835ffb9..b38758a 100644 --- a/src/main/java/lion/Externel/BackupSubServer.java +++ b/src/main/java/lion/Externel/BackupSubServer.java @@ -11,8 +11,7 @@ import java.net.Socket; import java.net.URLDecoder; import java.nio.charset.StandardCharsets; import java.util.*; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.*; /** HTTP distributor for the last-known-good subscription snapshot. */ @Slf4j @@ -24,17 +23,24 @@ public final class BackupSubServer implements Runnable { public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) { this.snapshotStore = Objects.requireNonNull(snapshotStore); this.port = port; - this.workers = Executors.newFixedThreadPool(Math.max(1, workerCount)); + int workersCount = Math.max(1, workerCount); + this.workers = new ThreadPoolExecutor(workersCount, workersCount, 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(workersCount * 32), new ThreadPoolExecutor.AbortPolicy()); } @Override public void run() { - snapshotStore.load(); try (ServerSocket serverSocket = new ServerSocket(port)) { log.info("备机订阅服务监听端口 {}", port); while (!Thread.currentThread().isInterrupted()) { Socket socket = serverSocket.accept(); - workers.execute(() -> handle(socket)); + try { + workers.execute(() -> handle(socket)); + } catch (RejectedExecutionException e) { + try (socket) { + send(socket, 503, "Service Unavailable", "text/plain", new byte[0], false); + } catch (IOException ignored) { } + } } } catch (IOException e) { log.error("备机订阅服务停止: {}", e.getMessage()); diff --git a/src/main/java/lion/Main.java b/src/main/java/lion/Main.java index 41c631f..412f152 100644 --- a/src/main/java/lion/Main.java +++ b/src/main/java/lion/Main.java @@ -15,6 +15,7 @@ public class Main { SubscriptionSnapshotStore snapshotStore = new SubscriptionSnapshotStore( java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret, Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes); + snapshotStore.load(); new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers), "subscription-backup-http").start(); new Thread(() -> MultiThreadedHTTPServer.main(null)).start(); diff --git a/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java b/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java index c9b8c88..5ed7fce 100644 --- a/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java +++ b/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java @@ -2,10 +2,12 @@ package lion.Message.Main; import lombok.Data; import lombok.NoArgsConstructor; +import lombok.ToString; import lion.Message.AbstractMessage; @Data @NoArgsConstructor +@ToString(exclude = {"payloadBase64", "signature"}) public class SubscriptionSnapshotMessage extends AbstractMessage { { messageType = SUBSCRIPTION_SNAPSHOT_MESSAGE; diff --git a/src/main/java/lion/Service/SubscriptionSnapshotStore.java b/src/main/java/lion/Service/SubscriptionSnapshotStore.java index 47e125f..caa3862 100644 --- a/src/main/java/lion/Service/SubscriptionSnapshotStore.java +++ b/src/main/java/lion/Service/SubscriptionSnapshotStore.java @@ -46,35 +46,47 @@ public final class SubscriptionSnapshotStore { } public void load() { + List candidates = new ArrayList<>(); try { Path pointer = root.resolve("current-revision"); - if (!Files.isRegularFile(pointer)) { - log.warn("没有找到订阅快照，备机订阅暂不可用"); - return; + if (Files.isRegularFile(pointer)) { + String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim(); + if (isRevision(revision)) candidates.add(revision); + } + Path snapshots = root.resolve("snapshots"); + if (Files.isDirectory(snapshots)) { + try (var stream = Files.list(snapshots)) { + stream.filter(Files::isDirectory) + .map(path -> path.getFileName().toString()) + .filter(SubscriptionSnapshotStore::isRevision) + .filter(revision -> !candidates.contains(revision)) + .sorted(Comparator.reverseOrder()) + .forEach(candidates::add); + } } - String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim(); - if (!isRevision(revision)) - throw new IOException("当前快照 revision 非法"); - Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision)); - current.set(snapshot); - log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount()); } catch (Exception e) { - current.set(null); - log.error("加载订阅快照失败，备机订阅暂不可用: {}", e.getMessage()); + log.warn("扫描订阅快照失败: {}", e.getMessage()); } + for (String revision : candidates) { + try { + Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision)); + current.set(snapshot); + writePointer(revision); + log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount()); + return; + } catch (Exception e) { + log.warn("订阅快照损坏，尝试上一版本 revision={}: {}", shortRevision(revision), e.getMessage()); + } + } + current.set(null); + log.warn("没有可用的订阅快照，备机订阅暂不可用"); } public ApplyResult apply(SubscriptionSnapshotMessage message) { + Path staging = null; try { if (message == null || message.getSchemaVersion() != 1 || !isRevision(message.getRevision())) return new ApplyResult(APPLY_INVALID, "消息版本或 revision 非法"); - Snapshot old = current.get(); - if (old != null) { - if (message.getRevision().equals(old.revision())) - return new ApplyResult(APPLY_OLD, "revision 已存在"); - if (message.getGeneratedAt() < old.generatedAt()) - return new ApplyResult(APPLY_OLD, "快照时间早于当前版本"); - } if (syncSecret.length == 0) return new ApplyResult(APPLY_INVALID, "同步密钥未配置"); @@ -91,10 +103,17 @@ public final class SubscriptionSnapshotStore { return new ApplyResult(APPLY_INVALID, "revision 与 payload 不一致"); SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class); SnapshotData data = validatePayload(payload); + Snapshot old = current.get(); + if (old != null) { + if (message.getRevision().equals(old.revision())) + return new ApplyResult(APPLY_OLD, "revision 已存在"); + if (message.getGeneratedAt() < old.generatedAt()) + return new ApplyResult(APPLY_OLD, "快照时间早于当前版本"); + } Path snapshots = root.resolve("snapshots"); Files.createDirectories(snapshots); - Path staging = snapshots.resolve(".staging-" + message.getRevision()); + staging = snapshots.resolve(".staging-" + message.getRevision()); deleteRecursively(staging); Files.createDirectories(staging.resolve("accounts")); for (AccountFiles account : data.accounts.values()) { @@ -106,17 +125,15 @@ public final class SubscriptionSnapshotStore { Map manifest = new LinkedHashMap<>(); manifest.put("revision", message.getRevision()); manifest.put("generatedAt", message.getGeneratedAt()); - manifest.put("payload", payload); + // 保留主站签名对应的原始 JSON 字节，重启时不依赖 Jackson 再序列化顺序。 + manifest.put("payloadBase64", Base64.getEncoder().encodeToString(payloadBytes)); Files.write(staging.resolve("manifest.json"), objectMapper.writeValueAsBytes(manifest), StandardOpenOption.CREATE_NEW); Path destination = snapshots.resolve(message.getRevision()); if (Files.exists(destination)) deleteRecursively(destination); atomicMove(staging, destination); - Path pointerTmp = root.resolve("current-revision.tmp"); - Files.writeString(pointerTmp, message.getRevision() + "\n", StandardCharsets.UTF_8, - StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING); - atomicMove(pointerTmp, root.resolve("current-revision")); + writePointer(message.getRevision()); Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash, data.accounts.size(), data.bindingCount); @@ -126,9 +143,22 @@ public final class SubscriptionSnapshotStore { } catch (Exception e) { log.error("应用订阅快照失败: {}", e.getMessage()); return new ApplyResult(APPLY_IO_ERROR, e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage()); + } finally { + if (staging != null && Files.exists(staging)) { + try { deleteRecursively(staging); } + catch (IOException e) { log.warn("清理订阅快照 staging 失败: {}", staging); } + } } } + private void writePointer(String revision) throws IOException { + Files.createDirectories(root); + Path pointerTmp = root.resolve("current-revision.tmp"); + Files.writeString(pointerTmp, revision + "\n", StandardCharsets.UTF_8, + StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING); + atomicMove(pointerTmp, root.resolve("current-revision")); + } + public Lookup lookup(String client, String publicKey) { Snapshot snapshot = current.get(); if (snapshot == null || snapshot.expired(System.currentTimeMillis(), maxStaleMillis)) @@ -152,9 +182,10 @@ public final class SubscriptionSnapshotStore { JsonNode manifest = objectMapper.readTree(Files.readAllBytes(directory.resolve("manifest.json"))); String revision = manifest.path("revision").asText(); long generatedAt = manifest.path("generatedAt").asLong(0); - SubscriptionSnapshotPayload payload = objectMapper.treeToValue(manifest.path("payload"), SubscriptionSnapshotPayload.class); + byte[] payloadBytes = decodeBase64(manifest.path("payloadBase64").asText(null), maxPayloadBytes); + SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class); SnapshotData data = validatePayload(payload); - if (!isRevision(revision) || !revision.equals(sha256(objectMapper.writeValueAsBytes(payload)))) + if (!isRevision(revision) || !constantEquals(revision, sha256(payloadBytes))) throw new IOException("快照 manifest revision 校验失败"); Map accounts = new HashMap<>(); for (SubscriptionAccountSnapshot account : payload.getAccounts()) { @@ -203,7 +234,10 @@ public final class SubscriptionSnapshotStore { Path snapshots = root.resolve("snapshots"); List dirs; try (var stream = Files.list(snapshots)) { - dirs = stream.filter(Files::isDirectory).filter(p -> !p.getFileName().toString().startsWith(".staging-")).toList(); + dirs = stream.filter(Files::isDirectory) + .filter(p -> !p.getFileName().toString().startsWith(".staging-")) + .sorted(Comparator.comparingLong(SubscriptionSnapshotStore::lastModified).reversed()) + .toList(); } dirs.stream().filter(p -> !p.getFileName().toString().equals(currentRevision)) .skip(1).forEach(p -> { @@ -264,6 +298,11 @@ public final class SubscriptionSnapshotStore { private static boolean isBase64Sha(String value) { return value != null && value.matches("[0-9a-fA-F]{64}"); } private static String shortRevision(String revision) { return revision == null ? null : revision.substring(0, Math.min(12, revision.length())); } + private static long lastModified(Path path) { + try { return Files.getLastModifiedTime(path).toMillis(); } + catch (IOException e) { return 0; } + } + private static void atomicMove(Path source, Path target) throws IOException { try { Files.move(source, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); } catch (AtomicMoveNotSupportedException e) { Files.move(source, target, StandardCopyOption.REPLACE_EXISTING); } diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java index 526a61e..ffbd9b7 100644 --- a/src/main/java/lion/storageNode.java +++ b/src/main/java/lion/storageNode.java @@ -1,5 +1,6 @@ package lion; +import lion.Config.Config; import lion.Domain.GalleryTask; import lion.Message.*; import lion.Message.Main.*; @@ -150,7 +151,10 @@ public class storageNode { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { - log.info(String.valueOf(msg)); + if (msg instanceof SubscriptionSnapshotMessage snapshot) + log.info("收到订阅快照 revision={}", shortRevision(snapshot.getRevision())); + else + log.info(String.valueOf(msg)); AbstractMessage abstractMessage = (AbstractMessage) msg; switch (abstractMessage.messageType){ @@ -201,7 +205,7 @@ public class storageNode { } case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> { SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage; - if (!ctx.channel().equals(server)) { + if (!Config.subscriptionSyncEnabled || !ctx.channel().equals(server)) { ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, SubscriptionSnapshotStore.APPLY_INVALID)); return; } diff --git a/src/main/resources/reflect-config.json b/src/main/resources/reflect-config.json index 86bc0f6..25d4ec6 100644 --- a/src/main/resources/reflect-config.json +++ b/src/main/resources/reflect-config.json @@ -62,6 +62,51 @@ "allDeclaredFields" : true, "allPublicFields" : true }, + { + "name": "lion.Message.Main.SubscriptionSnapshotMessage", + "allDeclaredConstructors" : true, + "allPublicConstructors" : true, + "allDeclaredMethods" : true, + "allPublicMethods" : true, + "allDeclaredFields" : true, + "allPublicFields" : true + }, + { + "name": "lion.Message.Main.SubscriptionSnapshotPayload", + "allDeclaredConstructors" : true, + "allPublicConstructors" : true, + "allDeclaredMethods" : true, + "allPublicMethods" : true, + "allDeclaredFields" : true, + "allPublicFields" : true + }, + { + "name": "lion.Message.Main.SubscriptionAccountSnapshot", + "allDeclaredConstructors" : true, + "allPublicConstructors" : true, + "allDeclaredMethods" : true, + "allPublicMethods" : true, + "allDeclaredFields" : true, + "allPublicFields" : true + }, + { + "name": "lion.Message.Main.SubscriptionBindingSnapshot", + "allDeclaredConstructors" : true, + "allPublicConstructors" : true, + "allDeclaredMethods" : true, + "allPublicMethods" : true, + "allDeclaredFields" : true, + "allPublicFields" : true + }, + { + "name": "lion.Service.SubscriptionSnapshotStore$Status", + "allDeclaredConstructors" : true, + "allPublicConstructors" : true, + "allDeclaredMethods" : true, + "allPublicMethods" : true, + "allDeclaredFields" : true, + "allPublicFields" : true + }, { "name": "lion.Message.AbstractMessage", "allDeclaredConstructors" : true, @@ -80,4 +125,4 @@ "allDeclaredFields" : true, "allPublicFields" : true } -] \ No newline at end of file +] diff --git a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java index 215d637..006cc0d 100644 --- a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java +++ b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java @@ -28,7 +28,7 @@ class SubscriptionSnapshotStoreTest { @Test void appliesSnapshotAndServesByHashedPublicKey(@TempDir Path directory) throws Exception { SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024); - SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000); + SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis()); assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code()); SubscriptionSnapshotStore.Lookup v2 = store.lookup("v2", "public-key-1"); @@ -41,7 +41,7 @@ class SubscriptionSnapshotStoreTest { @Test void rejectsTamperedPayloadAndKeepsPreviousSnapshot(@TempDir Path directory) throws Exception { SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024); - SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000); + SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis()); assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code()); message.setPayloadBase64(Base64.getEncoder().encodeToString("tampered".getBytes(StandardCharsets.UTF_8))); assertEquals(SubscriptionSnapshotStore.APPLY_INVALID, store.apply(message).code()); @@ -50,7 +50,7 @@ class SubscriptionSnapshotStoreTest { @Test void loadsLastGoodSnapshotAfterRestart(@TempDir Path directory) throws Exception { - SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000); + SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis()); SubscriptionSnapshotStore first = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024); assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, first.apply(message).code()); SubscriptionSnapshotStore second = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);