diff --git a/pom.xml b/pom.xml index 4c89862..284a1ec 100644 --- a/pom.xml +++ b/pom.xml @@ -64,10 +64,22 @@ httpclient 4.5.14 + + + org.junit.jupiter + junit-jupiter + 5.10.2 + test + + + org.apache.maven.plugins + maven-surefire-plugin + 3.2.5 + org.apache.maven.plugins maven-compiler-plugin @@ -128,4 +140,4 @@ - \ No newline at end of file + diff --git a/src/main/java/lion/Config/Config.java b/src/main/java/lion/Config/Config.java index 441c055..43caaf0 100644 --- a/src/main/java/lion/Config/Config.java +++ b/src/main/java/lion/Config/Config.java @@ -9,18 +9,37 @@ import lombok.extern.slf4j.Slf4j; @Slf4j public class Config { - public static String DouNaiV2ray; - public static String DouNaiClash; + public static boolean subscriptionSyncEnabled; + public static String subscriptionSyncSecret; + public static String subscriptionDataDir; + public static long subscriptionMaxStaleSeconds; + public static int subscriptionMaxPayloadBytes; + public static int subscriptionHttpPort; + public static int subscriptionHttpWorkers; + public static int subscriptionSocketTimeoutMs; public static void loadConfig(){ Properties prop = new Properties(); try (InputStream input = new FileInputStream("/root/gallery/storageNode/config.properties")) { prop.load(input); - DouNaiV2ray = prop.getProperty("DouNaiV2ray"); - DouNaiClash = prop.getProperty("DouNaiClash"); + subscriptionSyncEnabled = Boolean.parseBoolean(value(prop, "SubscriptionSyncEnabled", "false")); + subscriptionSyncSecret = System.getenv().getOrDefault("SUBSCRIPTION_SYNC_SECRET", + value(prop, "SubscriptionSyncSecret", "")); + subscriptionDataDir = value(prop, "SubscriptionDataDir", "/root/gallery/storageNode/sub"); + subscriptionMaxStaleSeconds = Long.parseLong(value(prop, "SubscriptionMaxStaleSeconds", "604800")); + subscriptionMaxPayloadBytes = Integer.parseInt(value(prop, "SubscriptionMaxPayloadBytes", "52428800")); + subscriptionHttpPort = Integer.parseInt(value(prop, "SubscriptionHttpPort", "8889")); + subscriptionHttpWorkers = Integer.parseInt(value(prop, "SubscriptionHttpWorkers", "4")); + subscriptionSocketTimeoutMs = Integer.parseInt(value(prop, "SubscriptionSocketTimeoutMs", "10000")); + if (subscriptionSyncEnabled && subscriptionSyncSecret.isBlank()) + throw new IllegalStateException("启用订阅同步时必须配置 SUBSCRIPTION_SYNC_SECRET"); } catch (IOException ex) { log.error("加载配置失败:{}", ex.getMessage()); } } + + private static String value(Properties prop, String key, String fallback) { + return prop.getProperty(key, fallback).trim(); + } } diff --git a/src/main/java/lion/Externel/BackupSubServer.java b/src/main/java/lion/Externel/BackupSubServer.java index 6246377..835ffb9 100644 --- a/src/main/java/lion/Externel/BackupSubServer.java +++ b/src/main/java/lion/Externel/BackupSubServer.java @@ -1,235 +1,153 @@ package lion.Externel; +import lion.Config.Config; +import lion.CustomUtil; +import lion.Service.SubscriptionSnapshotStore; import lombok.extern.slf4j.Slf4j; -import org.apache.http.HttpEntity; -import org.apache.http.client.methods.CloseableHttpResponse; -import org.apache.http.client.methods.HttpGet; -import org.apache.http.impl.client.CloseableHttpClient; -import org.apache.http.impl.client.HttpClients; import java.io.*; -import java.net.*; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.URLDecoder; import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.nio.file.Paths; import java.util.*; -import java.util.concurrent.*; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -import lion.CustomUtil; -import static lion.Config.Config.DouNaiClash; -import static lion.Config.Config.DouNaiV2ray; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +/** HTTP distributor for the last-known-good subscription snapshot. */ @Slf4j -public class BackupSubServer { +public final class BackupSubServer implements Runnable { + private final SubscriptionSnapshotStore snapshotStore; + private final int port; + private final ExecutorService workers; - public static void main(String[] args) { - updateSub(); - ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1); - threadPool.scheduleAtFixedRate(BackupSubServer::updateSub, 0, 12, TimeUnit.HOURS); + public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) { + this.snapshotStore = Objects.requireNonNull(snapshotStore); + this.port = port; + this.workers = Executors.newFixedThreadPool(Math.max(1, workerCount)); + } - String ip = ""; - try(ServerSocket serverSocket = new ServerSocket(8889)) { - log.info("Sub Server listening on port {}", 8889); - while (true) { - Socket clientSocket = serverSocket.accept(); - ip = clientSocket.getInetAddress().getHostAddress(); - log.info("Client connected:{}", ip); - handleClientRequest(clientSocket); + @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)); } } catch (IOException e) { - log.error("处理http请求时出错,IP:{},ERROR:{}", ip, e.getMessage()); + log.error("备机订阅服务停止: {}", e.getMessage()); + } finally { + workers.shutdownNow(); } } - public static void updateSub(){ - File DouNaiClashFile = new File("sub/DouNaiClash.txt"); - File DouNaiV2rayFile = new File("sub/DouNaiV2ray.txt"); - File directory = new File("sub"); - - if(!directory.isDirectory()) - try { - Files.createDirectory(Paths.get("sub")); - } catch (IOException e) { - log.error("create directory error:{}", e.getMessage()); - } - - List DouNaiClash_profile; - - //下载豆奶v2ray订阅 - try(FileWriter writer = new FileWriter(DouNaiV2rayFile)) { - String DouNaiV2rayRaw = Get(DouNaiV2ray).getFirst(); - String[] v2rayPlain = new String(Base64.getDecoder().decode(DouNaiV2rayRaw)).split("\n"); - StringBuilder stringBuilder = new StringBuilder(); - Pattern pattern = Pattern.compile("-?\\d+(\\.\\d+)?"); - - //过滤高倍率节点 - for(String node: v2rayPlain){ - String name = URLDecoder.decode(node.split("#")[1], StandardCharsets.UTF_8); - if(name.contains("流量")){ - Matcher matcher = pattern.matcher(name.substring(name.indexOf("(") + 1, name.indexOf(")"))); - - if (matcher.find()) { - float ratio = Float.parseFloat(matcher.group()); - if(ratio <= 2) { - stringBuilder.append(node).append("\n"); - continue; - } - } - stringBuilder.append(node).append("\n"); - } - else{ - stringBuilder.append(node).append("\n"); - } - } - writer.write(new String(Base64.getEncoder().encode(stringBuilder.toString().getBytes(StandardCharsets.UTF_8)))); - - log.info("load DouNai v2ray complete"); - }catch (IOException e){ - log.error("load DouNai v2ray failure: {}", e.getMessage()); - } - - //下载豆奶clash订阅 - try(FileWriter writer = new FileWriter(DouNaiClashFile)) { - DouNaiClash_profile = Get(DouNaiClash); - //过滤高倍率节点 - ArrayList clashProcessed = new ArrayList<>(); - boolean isProxies = false; - boolean skip = false; - for(String line: DouNaiClash_profile){ - if(line.equals("proxies:")) - isProxies = true; - else if(line.equals("proxy-groups:") && isProxies) - isProxies = false; - - if(isProxies) { - if (line.contains("name")) - skip = line.contains("流量"); - if (!skip) - clashProcessed.add(line); - } - else - if (!line.contains("流量")) - clashProcessed.add(line); - } - - for(String line: clashProcessed) - writer.write(line + "\n"); - - log.info("load DouNai clash complete"); - }catch (IOException e){ - log.error("load DouNai clash failure: {}", e.getMessage()); - } - } - - - public static ArrayList Get(String url) throws IOException { - try (CloseableHttpClient httpClient = HttpClients.createDefault()) { - HttpGet httpGet = new HttpGet(url); - try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) { - HttpEntity responseEntity = httpResponse.getEntity(); - int statusCode = httpResponse.getStatusLine().getStatusCode(); - ArrayList temp = new ArrayList<>(); - - if (statusCode == 200) { - BufferedReader reader = new BufferedReader(new InputStreamReader(responseEntity.getContent())); - String str; - while ((str = reader.readLine()) != null) - temp.add(str); - } - return temp; - } - } - } - - private static void handleClientRequest(Socket clientSocket) { - String fileName = ""; - try { - BufferedReader requestReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); - String requestLine = requestReader.readLine(); - - // Parse the request line to get the method and path - String[] requestParts = requestLine.split(" "); - String method = requestParts[0]; - Map paramMap = parseRequestLine(requestParts[1]);//path - if(paramMap == null){ - CustomUtil.sendErrorResponse(clientSocket, "404"); + private void handle(Socket socket) { + try (socket) { + socket.setSoTimeout(Config.subscriptionSocketTimeoutMs); + BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII)); + String requestLine = reader.readLine(); + if (requestLine == null || requestLine.length() > 2048) { + send(socket, 400, "Bad Request", "text/plain", new byte[0], false); return; } - log.info(Arrays.toString(requestParts)); - - // Only handle GET requests - if (method.equals("GET")) { - // Set the file path for download - File file = new File(fileName); - switch (paramMap.get("Client")) { - case "v2" -> file = new File("sub/DouNaiV2ray.txt"); - case "cat" -> file = new File("sub/DouNaiClash.txt"); - } - fileName = file.getName(); - log.info(file.getAbsolutePath()); - // Check if the file exists and is readable - if (file.exists() && file.isFile() && file.canRead()) { - // Get the file length - long fileLength = file.length(); - - // Get the range information for resuming download - long startByte = 0; - long endByte = fileLength - 1; - String rangeHeader = CustomUtil.getRequestHeader(requestReader); - if (rangeHeader != null && rangeHeader.startsWith("bytes=")) { - String[] rangeValues = rangeHeader.substring(6).split("-"); - startByte = Long.parseLong(rangeValues[0]); - if (rangeValues.length > 1 && !rangeValues[1].isEmpty()) { - endByte = Long.parseLong(rangeValues[1]); - } - } - - CustomUtil.sendFileRange(clientSocket, file, startByte, endByte); - } else { - // File not found or not readable, send 404 response - CustomUtil.sendErrorResponse(clientSocket, "404 Not Found"); - } - } else { - // Non-GET requests, send 501 response - CustomUtil.sendErrorResponse(clientSocket, "501 Not Implemented"); + String[] parts = requestLine.split(" ", 3); + if (parts.length != 3 || (!"GET".equals(parts[0]) && !"HEAD".equals(parts[0]))) { + send(socket, 405, "Method Not Allowed", "text/plain", new byte[0], false); + return; } - - // Close the request reader and client socket - requestReader.close(); - clientSocket.close(); - } catch (IOException e) { - log.error("处理文件下载时出错,IP:{}, 文件:{}, ERROR:{}", clientSocket.getInetAddress().getHostAddress(), fileName, e.getMessage()); + Map headers = readHeaders(reader); + if (headers == null) { + send(socket, 400, "Bad Request", "text/plain", new byte[0], false); + return; + } + String path = parts[1].split("\\?", 2)[0]; + if ("/health/subscription".equals(path)) { + byte[] body = CustomUtil.objectMapper.writeValueAsBytes(snapshotStore.status()); + send(socket, 200, "OK", "application/json; charset=utf-8", body, "HEAD".equals(parts[0])); + return; + } + String[] segments = path.split("/"); + if (segments.length != 4 || !"sub".equals(segments[1]) || !("v2".equals(segments[2]) || "cat".equals(segments[2]))) { + send(socket, 404, "Not Found", "text/plain", new byte[0], false); + return; + } + String key = URLDecoder.decode(segments[3], StandardCharsets.UTF_8); + if (key.length() < 6 || key.length() > 512) { + send(socket, 404, "Not Found", "text/plain", new byte[0], false); + return; + } + SubscriptionSnapshotStore.Lookup lookup = snapshotStore.lookup(segments[2], key); + if (lookup == null) { + SubscriptionSnapshotStore.Status status = snapshotStore.status(); + int code = "unavailable".equals(status.state()) || "expired".equals(status.state()) ? 503 : 404; + send(socket, code, code == 503 ? "Service Unavailable" : "Not Found", "text/plain", new byte[0], false); + return; + } + serveContent(socket, headers.get("range"), lookup.content(), "v2".equals(segments[2]), "HEAD".equals(parts[0])); + } catch (Exception e) { + log.debug("处理备机订阅请求失败: {}", e.getMessage()); } } - public static Map parseRequestLine(String requestLine) { - Map pathParams = new HashMap<>(); - - if(requestLine == null) - return null; - - String path; - if(requestLine.contains("?")) - path = requestLine.split("\\?")[0]; - else - path = requestLine; - - String[] vars = path.split("/"); - - if(vars.length < 4) - return null; - - pathParams.put("Key", vars[3]); - pathParams.put("Client", vars[2]); - - if(!pathParams.get("Client").equals("cat") && !pathParams.get("Client").equals("v2")) - return null; - - if(pathParams.get("Key").length()<6) - return null; - - return pathParams; + private static Map readHeaders(BufferedReader reader) throws IOException { + Map headers = new HashMap<>(); + int total = 0; + String line; + while ((line = reader.readLine()) != null) { + total += line.length(); + if (total > 8192) return null; + if (line.isEmpty()) return headers; + int colon = line.indexOf(':'); + if (colon <= 0) return null; + headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), line.substring(colon + 1).trim()); + } + return null; } -} \ No newline at end of file + + private static void serveContent(Socket socket, String range, byte[] content, boolean v2, boolean head) throws IOException { + long start = 0; + long end = content.length - 1L; + int status = 200; + String reason = "OK"; + if (range != null && range.startsWith("bytes=")) { + String value = range.substring(6).split(",", 2)[0]; + String[] values = value.split("-", 2); + try { + if (values.length != 2 || values[0].isEmpty()) throw new NumberFormatException(); + start = Long.parseLong(values[0]); + if (!values[1].isEmpty()) end = Long.parseLong(values[1]); + if (start < 0 || start > end || start >= content.length) throw new NumberFormatException(); + end = Math.min(end, content.length - 1L); + status = 206; + reason = "Partial Content"; + } catch (NumberFormatException e) { + send(socket, 416, "Range Not Satisfiable", v2 ? "text/plain" : "text/yaml", new byte[0], head); + return; + } + } + byte[] body = Arrays.copyOfRange(content, (int) start, (int) end + 1); + send(socket, status, reason, v2 ? "text/plain; charset=utf-8" : "text/yaml; charset=utf-8", body, head, + status == 206 ? "bytes " + start + "-" + end + "/" + content.length : null); + } + + private static void send(Socket socket, int status, String reason, String type, byte[] body, boolean head) throws IOException { + send(socket, status, reason, type, body, head, null); + } + + private static void send(Socket socket, int status, String reason, String type, byte[] body, boolean head, String range) throws IOException { + OutputStream output = socket.getOutputStream(); + StringBuilder header = new StringBuilder() + .append("HTTP/1.1 ").append(status).append(' ').append(reason).append("\r\n") + .append("Content-Type: ").append(type).append("\r\n") + .append("Content-Length: ").append(body.length).append("\r\n") + .append("Accept-Ranges: bytes\r\n") + .append("Connection: close\r\n"); + if (range != null) header.append("Content-Range: ").append(range).append("\r\n"); + header.append("\r\n"); + output.write(header.toString().getBytes(StandardCharsets.US_ASCII)); + if (!head) output.write(body); + output.flush(); + } +} diff --git a/src/main/java/lion/Main.java b/src/main/java/lion/Main.java index 235ab32..41c631f 100644 --- a/src/main/java/lion/Main.java +++ b/src/main/java/lion/Main.java @@ -3,6 +3,7 @@ package lion; import io.netty.bootstrap.Bootstrap; import lion.Config.Config; import lion.Externel.BackupSubServer; +import lion.Service.SubscriptionSnapshotStore; import lombok.extern.slf4j.Slf4j; @Slf4j @@ -11,9 +12,13 @@ public class Main { public static void main(String[] args) { boot(); Config.loadConfig(); - new Thread(() -> BackupSubServer.main(null)).start(); + SubscriptionSnapshotStore snapshotStore = new SubscriptionSnapshotStore( + java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret, + Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes); + new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers), + "subscription-backup-http").start(); new Thread(() -> MultiThreadedHTTPServer.main(null)).start(); - new storageNode(); + new storageNode(snapshotStore); } public static void boot(){ diff --git a/src/main/java/lion/Message/AbstractMessage.java b/src/main/java/lion/Message/AbstractMessage.java index 8a2633a..3fb6dd0 100644 --- a/src/main/java/lion/Message/AbstractMessage.java +++ b/src/main/java/lion/Message/AbstractMessage.java @@ -16,6 +16,8 @@ public class AbstractMessage { public static final byte AVAILABLE_CHECK_MESSAGE = 8; + public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9; + public byte messageType; public int messageId; diff --git a/src/main/java/lion/Message/Main/SubscriptionAccountSnapshot.java b/src/main/java/lion/Message/Main/SubscriptionAccountSnapshot.java new file mode 100644 index 0000000..cef5805 --- /dev/null +++ b/src/main/java/lion/Message/Main/SubscriptionAccountSnapshot.java @@ -0,0 +1,16 @@ +package lion.Message.Main; + +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@NoArgsConstructor +public class SubscriptionAccountSnapshot { + private Integer accountId; + private boolean enabled; + private boolean filterHighMultiplier; + private String v2ContentBase64; + private String v2Sha256; + private String clashContentBase64; + private String clashSha256; +} diff --git a/src/main/java/lion/Message/Main/SubscriptionBindingSnapshot.java b/src/main/java/lion/Message/Main/SubscriptionBindingSnapshot.java new file mode 100644 index 0000000..7ffe394 --- /dev/null +++ b/src/main/java/lion/Message/Main/SubscriptionBindingSnapshot.java @@ -0,0 +1,11 @@ +package lion.Message.Main; + +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@NoArgsConstructor +public class SubscriptionBindingSnapshot { + private String publicKeySha256; + private Integer accountId; +} diff --git a/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java b/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java new file mode 100644 index 0000000..c9b8c88 --- /dev/null +++ b/src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java @@ -0,0 +1,20 @@ +package lion.Message.Main; + +import lombok.Data; +import lombok.NoArgsConstructor; +import lion.Message.AbstractMessage; + +@Data +@NoArgsConstructor +public class SubscriptionSnapshotMessage extends AbstractMessage { + { + messageType = SUBSCRIPTION_SNAPSHOT_MESSAGE; + } + + private int schemaVersion; + private String revision; + private long generatedAt; + private String payloadBase64; + private String payloadSha256; + private String signature; +} diff --git a/src/main/java/lion/Message/Main/SubscriptionSnapshotPayload.java b/src/main/java/lion/Message/Main/SubscriptionSnapshotPayload.java new file mode 100644 index 0000000..d0b04f0 --- /dev/null +++ b/src/main/java/lion/Message/Main/SubscriptionSnapshotPayload.java @@ -0,0 +1,15 @@ +package lion.Message.Main; + +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.ArrayList; +import java.util.List; + +@Data +@NoArgsConstructor +public class SubscriptionSnapshotPayload { + private int schemaVersion; + private List accounts = new ArrayList<>(); + private List bindings = new ArrayList<>(); +} diff --git a/src/main/java/lion/Message/MessageCodec.java b/src/main/java/lion/Message/MessageCodec.java index d1f5689..0dc64e9 100644 --- a/src/main/java/lion/Message/MessageCodec.java +++ b/src/main/java/lion/Message/MessageCodec.java @@ -50,6 +50,7 @@ public class MessageCodec extends ByteToMessageCodec { case AbstractMessage.IDENTITY_MESSAGE -> objectMapper.readValue(metadata, IdentityMessage.class); case AbstractMessage.MAINTAIN_MESSAGE -> objectMapper.readValue(metadata, MaintainMessage.class); case AbstractMessage.AVAILABLE_CHECK_MESSAGE -> objectMapper.readValue(metadata, AvailableCheckMessage.class); + case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> objectMapper.readValue(metadata, SubscriptionSnapshotMessage.class); default -> null; }; diff --git a/src/main/java/lion/Service/SubscriptionSnapshotStore.java b/src/main/java/lion/Service/SubscriptionSnapshotStore.java new file mode 100644 index 0000000..47e125f --- /dev/null +++ b/src/main/java/lion/Service/SubscriptionSnapshotStore.java @@ -0,0 +1,289 @@ +package lion.Service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import lion.CustomUtil; +import lion.Message.Main.SubscriptionAccountSnapshot; +import lion.Message.Main.SubscriptionBindingSnapshot; +import lion.Message.Main.SubscriptionSnapshotMessage; +import lion.Message.Main.SubscriptionSnapshotPayload; +import lombok.extern.slf4j.Slf4j; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.*; +import java.security.MessageDigest; +import java.util.*; +import java.util.concurrent.atomic.AtomicReference; +import java.util.zip.GZIPInputStream; + +/** + * Last-known-good subscription data used by the standby HTTP server. + * The store never downloads upstream subscriptions and never stores upstream keys. + */ +@Slf4j +public final class SubscriptionSnapshotStore { + public static final byte APPLY_SUCCESS = 0; + public static final byte APPLY_INVALID = 1; + public static final byte APPLY_IO_ERROR = 2; + public static final byte APPLY_OLD = 3; + + private final Path root; + private final byte[] syncSecret; + private final long maxStaleMillis; + private final int maxPayloadBytes; + private final ObjectMapper objectMapper = CustomUtil.objectMapper; + private final AtomicReference current = new AtomicReference<>(); + + public SubscriptionSnapshotStore(Path root, String syncSecret, long maxStaleSeconds, int maxPayloadBytes) { + this.root = Objects.requireNonNull(root); + this.syncSecret = syncSecret == null ? new byte[0] : syncSecret.getBytes(StandardCharsets.UTF_8); + this.maxStaleMillis = Math.max(0, maxStaleSeconds) * 1000L; + this.maxPayloadBytes = maxPayloadBytes; + } + + public void load() { + try { + Path pointer = root.resolve("current-revision"); + if (!Files.isRegularFile(pointer)) { + log.warn("没有找到订阅快照，备机订阅暂不可用"); + return; + } + 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()); + } + } + + public ApplyResult apply(SubscriptionSnapshotMessage message) { + 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, "同步密钥未配置"); + + byte[] compressed = decodeBase64(message.getPayloadBase64(), maxPayloadBytes); + if (!constantEquals(message.getPayloadSha256(), sha256(compressed))) + return new ApplyResult(APPLY_INVALID, "payload SHA-256 校验失败"); + String signatureInput = message.getSchemaVersion() + "\n" + message.getRevision() + "\n" + + message.getGeneratedAt() + "\n" + message.getPayloadSha256(); + if (!constantEquals(message.getSignature(), hmac(signatureInput.getBytes(StandardCharsets.UTF_8)))) + return new ApplyResult(APPLY_INVALID, "快照签名校验失败"); + + byte[] payloadBytes = gunzip(compressed, maxPayloadBytes); + if (!constantEquals(message.getRevision(), sha256(payloadBytes))) + return new ApplyResult(APPLY_INVALID, "revision 与 payload 不一致"); + SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class); + SnapshotData data = validatePayload(payload); + + Path snapshots = root.resolve("snapshots"); + Files.createDirectories(snapshots); + Path staging = snapshots.resolve(".staging-" + message.getRevision()); + deleteRecursively(staging); + Files.createDirectories(staging.resolve("accounts")); + for (AccountFiles account : data.accounts.values()) { + Path dir = staging.resolve("accounts").resolve(String.valueOf(account.accountId())); + Files.createDirectories(dir); + Files.write(dir.resolve("v2ray.txt"), account.v2(), StandardOpenOption.CREATE_NEW); + Files.write(dir.resolve("clash.yaml"), account.clash(), StandardOpenOption.CREATE_NEW); + } + Map manifest = new LinkedHashMap<>(); + manifest.put("revision", message.getRevision()); + manifest.put("generatedAt", message.getGeneratedAt()); + manifest.put("payload", payload); + 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")); + + Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash, + data.accounts.size(), data.bindingCount); + current.set(snapshot); + cleanupOldSnapshots(message.getRevision()); + return new ApplyResult(APPLY_SUCCESS, "同步成功"); + } catch (Exception e) { + log.error("应用订阅快照失败: {}", e.getMessage()); + return new ApplyResult(APPLY_IO_ERROR, e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage()); + } + } + + public Lookup lookup(String client, String publicKey) { + Snapshot snapshot = current.get(); + if (snapshot == null || snapshot.expired(System.currentTimeMillis(), maxStaleMillis)) + return null; + AccountFiles account = snapshot.byKeyHash().get(sha256(publicKey.getBytes(StandardCharsets.UTF_8))); + if (account == null) + return null; + return new Lookup("v2".equals(client) ? account.v2() : account.clash(), snapshot.revision(), snapshot.generatedAt()); + } + + public Status status() { + Snapshot snapshot = current.get(); + if (snapshot == null) + return new Status("unavailable", null, 0, 0, 0); + long age = Math.max(0, System.currentTimeMillis() - snapshot.generatedAt()); + boolean expired = snapshot.expired(System.currentTimeMillis(), maxStaleMillis); + return new Status(expired ? "expired" : "ready", snapshot.revision(), snapshot.accountCount(), snapshot.bindingCount(), age); + } + + private Snapshot loadSnapshot(Path directory) throws IOException { + 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); + SnapshotData data = validatePayload(payload); + if (!isRevision(revision) || !revision.equals(sha256(objectMapper.writeValueAsBytes(payload)))) + throw new IOException("快照 manifest revision 校验失败"); + Map accounts = new HashMap<>(); + for (SubscriptionAccountSnapshot account : payload.getAccounts()) { + Path dir = directory.resolve("accounts").resolve(String.valueOf(account.getAccountId())); + byte[] v2 = Files.readAllBytes(dir.resolve("v2ray.txt")); + byte[] clash = Files.readAllBytes(dir.resolve("clash.yaml")); + if (!constantEquals(account.getV2Sha256(), sha256(v2)) || !constantEquals(account.getClashSha256(), sha256(clash))) + throw new IOException("缓存文件校验失败"); + accounts.put(account.getAccountId(), new AccountFiles(account.getAccountId(), v2, clash)); + } + Map byKey = new HashMap<>(); + for (SubscriptionBindingSnapshot binding : payload.getBindings()) + byKey.put(binding.getPublicKeySha256(), accounts.get(binding.getAccountId())); + return new Snapshot(revision, generatedAt, byKey, accounts.size(), payload.getBindings().size()); + } + + private SnapshotData validatePayload(SubscriptionSnapshotPayload payload) throws IOException { + if (payload == null || payload.getSchemaVersion() != 1 || payload.getAccounts() == null || payload.getBindings() == null) + throw new IOException("payload 版本或字段非法"); + if (payload.getAccounts().size() > 100 || payload.getBindings().size() > 10000) + throw new IOException("快照条目数量超限"); + Map accounts = new HashMap<>(); + for (SubscriptionAccountSnapshot account : payload.getAccounts()) { + if (account == null || account.getAccountId() == null || account.getAccountId() <= 0 || !account.isEnabled() + || !isBase64Sha(account.getV2Sha256()) || !isBase64Sha(account.getClashSha256())) + throw new IOException("账号字段非法"); + byte[] v2 = decodeBase64(account.getV2ContentBase64(), maxPayloadBytes); + byte[] clash = decodeBase64(account.getClashContentBase64(), maxPayloadBytes); + if (!constantEquals(account.getV2Sha256(), sha256(v2)) || !constantEquals(account.getClashSha256(), sha256(clash))) + throw new IOException("账号缓存 SHA-256 校验失败"); + if (accounts.put(account.getAccountId(), new AccountFiles(account.getAccountId(), v2, clash)) != null) + throw new IOException("账号 ID 重复"); + } + Map byKey = new HashMap<>(); + for (SubscriptionBindingSnapshot binding : payload.getBindings()) { + if (binding == null || !isBase64Sha(binding.getPublicKeySha256()) || !accounts.containsKey(binding.getAccountId())) + throw new IOException("绑定字段或账号引用非法"); + if (byKey.put(binding.getPublicKeySha256(), accounts.get(binding.getAccountId())) != null) + throw new IOException("公开 Key Hash 重复"); + } + return new SnapshotData(accounts, byKey, payload.getBindings().size()); + } + + private void cleanupOldSnapshots(String currentRevision) { + try { + 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(p -> !p.getFileName().toString().equals(currentRevision)) + .skip(1).forEach(p -> { + try { deleteRecursively(p); } catch (IOException e) { log.warn("清理旧订阅快照失败: {}", p); } + }); + } catch (IOException e) { + log.warn("扫描旧订阅快照失败: {}", e.getMessage()); + } + } + + private static byte[] gunzip(byte[] compressed, int limit) throws IOException { + try (GZIPInputStream input = new GZIPInputStream(new ByteArrayInputStream(compressed))) { + byte[] buffer = new byte[8192]; + var output = new java.io.ByteArrayOutputStream(); + int total = 0, read; + while ((read = input.read(buffer)) != -1) { + total += read; + if (total > limit) throw new IOException("解压后 payload 超限"); + output.write(buffer, 0, read); + } + return output.toByteArray(); + } + } + + private String hmac(byte[] input) throws Exception { + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(syncSecret, "HmacSHA256")); + return hex(mac.doFinal(input)); + } + + private static byte[] decodeBase64(String value, int maxBytes) throws IOException { + if (value == null || value.length() > maxBytes * 2L) + throw new IOException("Base64 数据超限"); + try { + byte[] result = Base64.getDecoder().decode(value); + if (result.length > maxBytes) throw new IOException("数据超限"); + return result; + } catch (IllegalArgumentException e) { + throw new IOException("Base64 数据非法"); + } + } + + private static String sha256(byte[] bytes) { + try { return hex(MessageDigest.getInstance("SHA-256").digest(bytes)); } + catch (Exception e) { throw new IllegalStateException(e); } + } + + private static String hex(byte[] bytes) { + return HexFormat.of().formatHex(bytes); + } + + private static boolean constantEquals(String expected, String actual) { + return expected != null && MessageDigest.isEqual(expected.toLowerCase(Locale.ROOT).getBytes(StandardCharsets.US_ASCII), + actual.toLowerCase(Locale.ROOT).getBytes(StandardCharsets.US_ASCII)); + } + + private static boolean isRevision(String value) { return isBase64Sha(value); } + 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 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); } + } + + private static void deleteRecursively(Path path) throws IOException { + if (!Files.exists(path)) return; + try (var stream = Files.walk(path)) { + stream.sorted(Comparator.reverseOrder()).forEach(p -> { + try { Files.deleteIfExists(p); } catch (IOException e) { throw new UncheckedIOException(e); } + }); + } catch (UncheckedIOException e) { throw e.getCause(); } + } + + public record ApplyResult(byte code, String message) {} + public record Lookup(byte[] content, String revision, long generatedAt) {} + public record Status(String state, String revision, int accountCount, int bindingCount, long ageMillis) {} + private record AccountFiles(Integer accountId, byte[] v2, byte[] clash) {} + private record SnapshotData(Map accounts, Map byKeyHash, int bindingCount) {} + private record Snapshot(String revision, long generatedAt, Map byKeyHash, int accountCount, int bindingCount) { + boolean expired(long now, long maxAge) { return maxAge > 0 && now - generatedAt > maxAge; } + } +} diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java index 215cd91..526a61e 100644 --- a/src/main/java/lion/storageNode.java +++ b/src/main/java/lion/storageNode.java @@ -4,6 +4,7 @@ import lion.Domain.GalleryTask; import lion.Message.*; import lion.Message.Main.*; import lion.Service.DeleteService; +import lion.Service.SubscriptionSnapshotStore; import lion.Service.DownloadCheckService; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; @@ -37,9 +38,19 @@ public class storageNode { ReentrantLock lock; + final SubscriptionSnapshotStore subscriptionSnapshotStore; + + final ExecutorService subscriptionApplyExecutor; + public static String storagePath = "/root/gallery/gallery/"; - public storageNode(){ + public storageNode(SubscriptionSnapshotStore subscriptionSnapshotStore){ + this.subscriptionSnapshotStore = subscriptionSnapshotStore; + this.subscriptionApplyExecutor = Executors.newSingleThreadExecutor(r -> { + Thread thread = new Thread(r, "subscription-snapshot-apply"); + thread.setDaemon(true); + return thread; + }); queue = new HashMap<>(); tempQueue = new HashMap<>(); lock = new ReentrantLock(); @@ -188,6 +199,18 @@ public class storageNode { ResponseMessage responseMessage = new ResponseMessage(acm.messageId, (byte)0); ctx.writeAndFlush(responseMessage); } + case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> { + SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage; + if (!ctx.channel().equals(server)) { + ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, SubscriptionSnapshotStore.APPLY_INVALID)); + return; + } + subscriptionApplyExecutor.execute(() -> { + SubscriptionSnapshotStore.ApplyResult result = subscriptionSnapshotStore.apply(snapshotMessage); + ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, result.code())); + log.info("订阅快照处理完成 revision={} result={}", shortRevision(snapshotMessage.getRevision()), result.code()); + }); + } } // // //修复预览 @@ -206,4 +229,8 @@ public class storageNode { } } } + + private static String shortRevision(String revision) { + return revision == null ? null : revision.substring(0, Math.min(12, revision.length())); + } } diff --git a/src/main/resources/config.properties b/src/main/resources/config.properties index 427b6b4..94f22ad 100644 --- a/src/main/resources/config.properties +++ b/src/main/resources/config.properties @@ -1,2 +1,9 @@ -DouNaiV2ray=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=v2 -DouNaiClash=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=clashmeta \ No newline at end of file +# 订阅快照由主站通过 Netty 长连接推送；不要在此保存上游订阅地址或 Key。 +SubscriptionSyncEnabled=false +SubscriptionSyncSecret= +SubscriptionDataDir=/root/gallery/storageNode/sub +SubscriptionMaxStaleSeconds=604800 +SubscriptionMaxPayloadBytes=52428800 +SubscriptionHttpPort=8889 +SubscriptionHttpWorkers=4 +SubscriptionSocketTimeoutMs=10000 diff --git a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java new file mode 100644 index 0000000..215d637 --- /dev/null +++ b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java @@ -0,0 +1,104 @@ +package lion.Service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lion.CustomUtil; +import lion.Message.Main.SubscriptionAccountSnapshot; +import lion.Message.Main.SubscriptionBindingSnapshot; +import lion.Message.Main.SubscriptionSnapshotMessage; +import lion.Message.Main.SubscriptionSnapshotPayload; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.util.Base64; +import java.util.HexFormat; +import java.util.zip.GZIPOutputStream; + +import static org.junit.jupiter.api.Assertions.*; + +class SubscriptionSnapshotStoreTest { + private static final String SECRET = "snapshot-test-secret"; + private final ObjectMapper mapper = CustomUtil.objectMapper; + + @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); + + assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code()); + SubscriptionSnapshotStore.Lookup v2 = store.lookup("v2", "public-key-1"); + SubscriptionSnapshotStore.Lookup clash = store.lookup("cat", "public-key-1"); + assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), v2.content()); + assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), clash.content()); + assertEquals(SubscriptionSnapshotStore.APPLY_OLD, store.apply(message).code()); + } + + @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); + 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()); + assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), store.lookup("v2", "public-key-1").content()); + } + + @Test + void loadsLastGoodSnapshotAfterRestart(@TempDir Path directory) throws Exception { + SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000); + 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); + second.load(); + assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), second.lookup("cat", "public-key-1").content()); + } + + private SubscriptionSnapshotMessage message(String publicKey, String v2, String clash, long generatedAt) throws Exception { + byte[] v2Bytes = v2.getBytes(StandardCharsets.UTF_8); + byte[] clashBytes = clash.getBytes(StandardCharsets.UTF_8); + SubscriptionAccountSnapshot account = new SubscriptionAccountSnapshot(); + account.setAccountId(1); + account.setEnabled(true); + account.setV2ContentBase64(Base64.getEncoder().encodeToString(v2Bytes)); + account.setV2Sha256(sha256(v2Bytes)); + account.setClashContentBase64(Base64.getEncoder().encodeToString(clashBytes)); + account.setClashSha256(sha256(clashBytes)); + SubscriptionBindingSnapshot binding = new SubscriptionBindingSnapshot(); + binding.setPublicKeySha256(sha256(publicKey.getBytes(StandardCharsets.UTF_8))); + binding.setAccountId(1); + SubscriptionSnapshotPayload payload = new SubscriptionSnapshotPayload(); + payload.setSchemaVersion(1); + payload.setAccounts(java.util.List.of(account)); + payload.setBindings(java.util.List.of(binding)); + byte[] json = mapper.writeValueAsBytes(payload); + byte[] compressed = gzip(json); + SubscriptionSnapshotMessage message = new SubscriptionSnapshotMessage(); + message.setSchemaVersion(1); + message.setRevision(sha256(json)); + message.setGeneratedAt(generatedAt); + message.setPayloadBase64(Base64.getEncoder().encodeToString(compressed)); + message.setPayloadSha256(sha256(compressed)); + String input = "1\n" + message.getRevision() + "\n" + generatedAt + "\n" + message.getPayloadSha256(); + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); + message.setSignature(hex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8)))); + return message; + } + + private static byte[] gzip(byte[] bytes) throws Exception { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (GZIPOutputStream gzip = new GZIPOutputStream(output)) { gzip.write(bytes); } + return output.toByteArray(); + } + + private static String sha256(byte[] bytes) throws Exception { + return hex(MessageDigest.getInstance("SHA-256").digest(bytes)); + } + + private static String hex(byte[] bytes) { return HexFormat.of().formatHex(bytes); } +}