diff --git a/llm_readme.txt b/llm_readme.txt index c5329d6..61942bc 100644 --- a/llm_readme.txt +++ b/llm_readme.txt @@ -53,7 +53,9 @@ storageNode/ │ ├── simplelogger.properties # SLF4J 日志配置（输出到 run.out） │ └── reflect-config.json # GraalVM 反射配置（Jackson 序列化） └── test/ - └── java/ # 订阅快照、下载/压缩恢复与主站通道自愈测试 + └── java/ # 协议编解码、节点消息处理与上报循环、 + # 订阅快照与分发、下载/压缩恢复、 + # 主站通道自愈、删除与工具方法测试 ``` --- @@ -170,7 +172,9 @@ SubscriptionSocketTimeoutMs=10000 ## 架构说明 -- 单模块 Maven 项目，现有 5 个测试用例覆盖订阅快照和压缩失败恢复 +- 单模块 Maven 项目，现有 62 个测试用例：协议编解码（MessageCodec）、节点消息处理与 + 状态上报循环、备机订阅分发 HTTP（8889）、订阅快照存取、下载/压缩恢复与 gid 匹配、 + 主站通道自愈、删除与工具方法 - 硬编码文件系统路径（`/root/gallery/...`）→ 仅限 Linux 部署 - 外部连接：`lionwebsite.xyz`、`personal.lionwebsite.xyz`、`aaaa.gay` - GraalVM 原生镜像编译，包含 Jackson 反射配置 diff --git a/src/main/java/lion/Config/Config.java b/src/main/java/lion/Config/Config.java index 43caaf0..41ac141 100644 --- a/src/main/java/lion/Config/Config.java +++ b/src/main/java/lion/Config/Config.java @@ -3,12 +3,16 @@ package lion.Config; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Properties; import lombok.extern.slf4j.Slf4j; @Slf4j public class Config { + public static final String CONFIG_PATH = "/root/gallery/storageNode/config.properties"; + public static boolean subscriptionSyncEnabled; public static String subscriptionSyncSecret; public static String subscriptionDataDir; @@ -19,24 +23,47 @@ public class Config { public static int subscriptionSocketTimeoutMs; public static void loadConfig(){ + loadConfig(CONFIG_PATH); + } + + /** + * 从指定文件加载配置。 + * + *

拆出带路径的版本是为了能直接对「默认值填充」和「启用同步却没配密钥就拒绝启动」 + * 写断言，不必依赖生产路径存在。文件缺失不是致命错误（保留内置默认值）， + * 但会记录明确日志，避免以错误配置静默运行。 + */ + public static void loadConfig(String path){ Properties prop = new Properties(); - try (InputStream input = new FileInputStream("/root/gallery/storageNode/config.properties")) { + if (!Files.isRegularFile(Path.of(path))) { + log.warn("配置文件不存在:{}，使用内置默认值", path); + apply(prop); + return; + } + + try (InputStream input = new FileInputStream(path)) { prop.load(input); - 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()); + apply(prop); + return; } + apply(prop); + } + + private static void apply(Properties prop){ + 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"); } private static String value(Properties prop, String key, String fallback) { diff --git a/src/main/java/lion/CustomUtil.java b/src/main/java/lion/CustomUtil.java index e65b940..da629cb 100644 --- a/src/main/java/lion/CustomUtil.java +++ b/src/main/java/lion/CustomUtil.java @@ -9,6 +9,8 @@ import java.io.*; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; @Slf4j public class CustomUtil { @@ -16,12 +18,15 @@ public class CustomUtil { public static ObjectMapper objectMapper = new ObjectMapper(); public static void notifyMe(String message) { - String url = "https://personal.lionwebsite.xyz/message2me?AuthCode=alone&message=" + message; - HttpRequest request = HttpRequest.post(url); + // 画廊名可能含空格、'&'、'#' 或非 ASCII，必须编码后再拼进查询串，否则会截断或串参数。 + String url = "https://personal.lionwebsite.xyz/message2me?AuthCode=alone&message=" + + URLEncoder.encode(message == null ? "" : message, StandardCharsets.UTF_8); + // 外呼必须有超时；调用方虽有专用守护线程，但不设超时仍可能长期占住它。 + HttpRequest request = HttpRequest.post(url).timeout(5_000); request.header("User-Agent", "Mozilla/5.0"); try(HttpResponse response = request.execute()) { if(response.getStatus() != 200) { - System.out.println("通知失败, status code:" + response.getStatus() + ", message:" + message); + log.warn("通知失败, status code:{}, message:{}", response.getStatus(), message); } } } @@ -65,6 +70,19 @@ public class CustomUtil { public static void sendFileRange(Socket clientSocket, File file, long startByte, long endByte, String contentDispositionFileName) throws IOException { long fileLength = file.length(); + + // 区间必须夹在文件范围内：客户端可能请求远超文件末端（例如 bytes=0-999999）， + // 若照原样写进 Content-Length / Content-Range，对端会一直等一个永不到来的长度， + // 连接要挂到超时才断。起点晚于终点（含空文件）则无可发送内容，直接回 416。 + if (startByte < 0) + startByte = 0; + if (endByte > fileLength - 1) + endByte = fileLength - 1; + if (startByte > endByte) { + sendErrorResponse(clientSocket, "416 Range Not Satisfiable"); + return; + } + OutputStream responseStream = clientSocket.getOutputStream(); PrintWriter responseWriter = new PrintWriter(responseStream, true); responseWriter.println("HTTP/1.1 206 Partial Content"); diff --git a/src/main/java/lion/Message/MessageCodec.java b/src/main/java/lion/Message/MessageCodec.java index 0dc64e9..902b6ab 100644 --- a/src/main/java/lion/Message/MessageCodec.java +++ b/src/main/java/lion/Message/MessageCodec.java @@ -5,6 +5,8 @@ import com.fasterxml.jackson.databind.ObjectMapper; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageCodec; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.EncoderException; import lion.CustomUtil; import lion.Message.Main.*; import lombok.extern.slf4j.Slf4j; @@ -15,6 +17,9 @@ import java.util.List; @Slf4j public class MessageCodec extends ByteToMessageCodec { + /** 单帧负载上限，与上游 LengthFieldBasedFrameDecoder 的 100MB 保持一致。 */ + private static final int MAX_FRAME_BYTES = 100_000_000; + ObjectMapper objectMapper; public MessageCodec(){ @@ -23,21 +28,25 @@ public class MessageCodec extends ByteToMessageCodec { @Override protected void encode(ChannelHandlerContext channelHandlerContext, AbstractMessage abstractMessage, ByteBuf byteBuf) { - byteBuf.writeByte(abstractMessage.messageType); - + // 先序列化、再写帧头：否则中途失败会留下只有类型字节、没有长度和 JSON 的半帧， + // 对端的分帧器会为了等长度前缀一直挂住这条连接。 + byte[] bytes; try { - byte[] bytes = objectMapper.writeValueAsBytes(abstractMessage); - byteBuf.writeInt(bytes.length); - byteBuf.writeBytes(bytes); + bytes = objectMapper.writeValueAsBytes(abstractMessage); } catch (Exception e) { - log.error("序列化消息失败:{}", e.getMessage()); + throw new EncoderException("序列化消息失败:" + e.getMessage(), e); } + byteBuf.writeByte(abstractMessage.messageType); + byteBuf.writeInt(bytes.length); + byteBuf.writeBytes(bytes); } @Override protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List list) throws Exception { byte messageType = byteBuf.readByte(); int length = byteBuf.readInt(); + if (length < 0 || length > MAX_FRAME_BYTES) + throw new CorruptedFrameException("非法的帧长度: " + length); byte[] bytes = new byte[length]; byteBuf.readBytes(bytes); final String metadata = new String(bytes, StandardCharsets.UTF_8); diff --git a/src/main/java/lion/MultiThreadedHTTPServer.java b/src/main/java/lion/MultiThreadedHTTPServer.java index e5ea61a..bf3f7e9 100644 --- a/src/main/java/lion/MultiThreadedHTTPServer.java +++ b/src/main/java/lion/MultiThreadedHTTPServer.java @@ -15,6 +15,12 @@ import java.util.concurrent.Executors; public class MultiThreadedHTTPServer { private static final int PORT = 8888; + /** 已安装的客户端可能连上却不发请求；没有读超时就会一直占住工作线程。 */ + private static final int SOCKET_READ_TIMEOUT_MILLIS = 30_000; + + /** 兜底用的不存在路径：让「按文件名找不到归档」走统一的 404 分支。 */ + private static final File FILE_NOT_FOUND = new File("/root/abc"); + public static void main(String[] args) { ExecutorService threadPool = Executors.newCachedThreadPool(); String real_ip; @@ -23,38 +29,54 @@ public class MultiThreadedHTTPServer { } catch (UnknownHostException ignored){ real_ip = "207.60.50.74"; } - try(ServerSocket serverSocket = new ServerSocket(PORT)) { - log.info("Server listening on port {}", PORT); - while (true) { - Socket clientSocket = serverSocket.accept(); - String ip = clientSocket.getInetAddress().getHostAddress(); - if(ip.equals(real_ip)){ - log.info("Client connected"); - threadPool.submit(() -> handleClientRequest(clientSocket)); - }else{ - log.info("unknown ip: " + ip); - clientSocket.close(); + // accept 循环必须扛得住单次失败：否则一次异常就让下载服务整体退出， + // 之后所有下载都 404，直到有人手工重启。 + while (!Thread.currentThread().isInterrupted()) { + try (ServerSocket serverSocket = new ServerSocket(PORT)) { + log.info("Server listening on port {}", PORT); + while (true) { + Socket clientSocket = serverSocket.accept(); + String ip = clientSocket.getInetAddress().getHostAddress(); + if(ip.equals(real_ip)){ + log.info("Client connected"); + threadPool.submit(() -> handleClientRequest(clientSocket)); + }else{ + log.info("unknown ip: " + ip); + clientSocket.close(); + } + } + } catch (IOException e) { + log.error("处理http请求时出错,IP:{},ERROR:{}", real_ip, e.getMessage()); + try { + Thread.sleep(5_000); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); } } - } catch (IOException e) { - log.error("处理http请求时出错,IP:{},ERROR:{}", real_ip, e.getMessage()); } } - private static void handleClientRequest(Socket clientSocket) { + // 包级可见：与 parseRequestLine 一样留作测试入口，便于对真实 socket 断言请求处理。 + static void handleClientRequest(Socket clientSocket) { String fileName = ""; - try { + try (Socket socket = clientSocket) { + // 没有读超时的连接可以永远占住一个工作线程，必须设一个上限。 + socket.setSoTimeout(SOCKET_READ_TIMEOUT_MILLIS); BufferedReader requestReader = new BufferedReader( - new InputStreamReader(clientSocket.getInputStream(), StandardCharsets.ISO_8859_1)); + new InputStreamReader(socket.getInputStream(), StandardCharsets.ISO_8859_1)); String requestLine = requestReader.readLine(); - if (requestLine == null) { - clientSocket.close(); + // 空行或没有空格的请求行会让 requestParts[1] 越界。这种异常不是 IOException， + // 原样抛出会被线程池静默吞掉、socket 也关不掉，因此在这里显式判非法。 + if (requestLine == null || requestLine.isBlank()) return; - } // Parse the request line to get the method and path - String[] requestParts = requestLine.split(" "); + String[] requestParts = requestLine.trim().split("\\s+"); + if (requestParts.length < 2) { + CustomUtil.sendErrorResponse(socket, "400 Bad Request"); + return; + } String method = requestParts[0]; Map paramMap = parseRequestLine(requestParts[1]);//path log.info(Arrays.toString(requestParts)); @@ -81,12 +103,12 @@ public class MultiThreadedHTTPServer { name = filePath + name + "/" + name + ".zip"; file = new File(name); }else{ - file = new File("/root/abc"); + file = FILE_NOT_FOUND; } } } else{ - CustomUtil.sendErrorResponse(clientSocket, "403 Forbidden"); + CustomUtil.sendErrorResponse(socket, "403 Forbidden"); return; } fileName = file.getName(); @@ -101,27 +123,36 @@ public class MultiThreadedHTTPServer { 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]); + // 非法 Range 不得中断处理：解析失败退回整文件， + // 区间越界则由 sendFileRange 统一夹取或回 416。 + String value = rangeHeader.substring(6).split(",", 2)[0]; + String[] rangeValues = value.split("-", 2); + try { + if (rangeValues[0].isEmpty()) + throw new NumberFormatException(); + startByte = Long.parseLong(rangeValues[0]); + if (rangeValues.length > 1 && !rangeValues[1].isEmpty()) + endByte = Long.parseLong(rangeValues[1]); + } catch (NumberFormatException invalidRange) { + startByte = 0; + endByte = fileLength - 1; } } - CustomUtil.sendFileRange(clientSocket, file, startByte, endByte, fileName); + CustomUtil.sendFileRange(socket, file, startByte, endByte, fileName); } else { // File not found or not readable, send 404 response - CustomUtil.sendErrorResponse(clientSocket, "404 Not Found"); + CustomUtil.sendErrorResponse(socket, "404 Not Found"); } } else { // Non-GET requests, send 501 response - CustomUtil.sendErrorResponse(clientSocket, "501 Not Implemented"); + CustomUtil.sendErrorResponse(socket, "501 Not Implemented"); } - // Close the request reader and client socket + // socket 由 try-with-resources 关闭，这里只收尾 reader。 requestReader.close(); - clientSocket.close(); - } catch (IOException e) { + } catch (Exception e) { + // 不能只捕 IOException：越界等运行时异常会绕过 finally，导致连接泄漏。 log.error("处理文件下载时出错,IP:{}, 文件:{}, ERROR:{}", clientSocket.getInetAddress().getHostAddress(), fileName, e.getMessage()); } } diff --git a/src/main/java/lion/Service/DeleteService.java b/src/main/java/lion/Service/DeleteService.java index aeff217..6e08cfd 100644 --- a/src/main/java/lion/Service/DeleteService.java +++ b/src/main/java/lion/Service/DeleteService.java @@ -4,6 +4,7 @@ import lion.ErrorCode.ErrorCode; import cn.hutool.core.io.FileUtil; import java.io.File; +import java.nio.file.Path; public class DeleteService { public static byte deleteAll(String path){ @@ -16,4 +17,23 @@ public class DeleteService { else return ErrorCode.IO_ERROR; } + + /** + * 删除某个根目录下的子目录，且保证不越界。 + * + *

画廊名来自主站下发的消息，直接拼接根路径会让 {@code ../} 或绝对路径 + * 操纵到根目录之外（例如删掉节点上任意目录）。这里按规范化后的路径判断包含关系， + * 越界与空名一律按「找不到」处理，绝不落到磁盘删除。 + */ + public static byte deleteWithin(String rootPath, String galleryName){ + if (galleryName == null || galleryName.isBlank()) + return ErrorCode.FILE_NOT_FOUND; + + Path root = new File(rootPath).toPath().toAbsolutePath().normalize(); + Path target = root.resolve(galleryName).normalize(); + if (target.equals(root) || !target.startsWith(root)) + return ErrorCode.FILE_NOT_FOUND; + + return deleteAll(target.toString()); + } } diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java index 90ac98b..c250774 100644 --- a/src/main/java/lion/storageNode.java +++ b/src/main/java/lion/storageNode.java @@ -76,7 +76,7 @@ public class storageNode { int real_port = CustomUtil._findIdlePort(26321); - new ServerBootstrap() + ChannelFuture bindFuture = new ServerBootstrap() .channel(NioServerSocketChannel.class) .group(new NioEventLoopGroup()) .childHandler(new ChannelInitializer() { @@ -87,7 +87,17 @@ public class storageNode { channel.pipeline().addLast(new MyChannelInboundHandlerAdapter(tempQueue)); } }).bind(real_port); - log.info("listening on port {}", real_port); + // 必须等绑定结果：端口被别人抢到（探测与绑定之间的 TOCTOU）时，若不检查， + // 节点会「启动成功」却根本没在监听，主站怎么连都连不上。 + try { + bindFuture.sync(); + log.info("listening on port {}", real_port); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("绑定端口 " + real_port + " 被中断", e); + } catch (Exception e) { + throw new IllegalStateException("绑定端口 " + real_port + " 失败", e); + } wakeMainServer(); downloadCheckService = new DownloadCheckService(queue); @@ -301,7 +311,7 @@ public class storageNode { } finally { lock.unlock(); } - byte result = DeleteService.deleteAll(storagePath + galleryName); + byte result = DeleteService.deleteWithin(storagePath, galleryName); ResponseMessage responseMessage = new ResponseMessage(deleteGalleryMessage.messageId, result); ctx.writeAndFlush(responseMessage); } diff --git a/src/test/java/lion/ConfigTest.java b/src/test/java/lion/ConfigTest.java new file mode 100644 index 0000000..3434570 --- /dev/null +++ b/src/test/java/lion/ConfigTest.java @@ -0,0 +1,82 @@ +package lion; + +import lion.Config.Config; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 配置加载。 + * + *

生产配置来自 {@code /root/gallery/storageNode/config.properties}，任何一项读错都会改变 + * 订阅同步与分发端口的行为。这里对默认值填充、显式覆盖、文件缺失回退与「启用同步却 + * 没配密钥就拒绝启动」逐条断言。 + */ +class ConfigTest { + + @Test + void fillsBuiltInDefaultsWhenKeysAreAbsent(@TempDir Path root) throws Exception { + Path file = Files.writeString(root.resolve("config.properties"), ""); + + Config.loadConfig(file.toString()); + + assertFalse(Config.subscriptionSyncEnabled); + assertEquals("/root/gallery/storageNode/sub", Config.subscriptionDataDir); + assertEquals(604800L, Config.subscriptionMaxStaleSeconds); + assertEquals(52428800, Config.subscriptionMaxPayloadBytes); + assertEquals(8889, Config.subscriptionHttpPort); + assertEquals(4, Config.subscriptionHttpWorkers); + assertEquals(10000, Config.subscriptionSocketTimeoutMs); + } + + @Test + void readsExplicitValues(@TempDir Path root) throws Exception { + Path file = Files.writeString(root.resolve("config.properties"), String.join("\n", + "SubscriptionSyncEnabled=true", + "SubscriptionSyncSecret=file-secret", + "SubscriptionDataDir=/tmp/sub", + "SubscriptionMaxStaleSeconds=120", + "SubscriptionMaxPayloadBytes=4096", + "SubscriptionHttpPort=9999", + "SubscriptionHttpWorkers=2", + "SubscriptionSocketTimeoutMs=500")); + + Config.loadConfig(file.toString()); + + assertTrue(Config.subscriptionSyncEnabled); + assertEquals("file-secret", Config.subscriptionSyncSecret); + assertEquals("/tmp/sub", Config.subscriptionDataDir); + assertEquals(120L, Config.subscriptionMaxStaleSeconds); + assertEquals(4096, Config.subscriptionMaxPayloadBytes); + assertEquals(9999, Config.subscriptionHttpPort); + assertEquals(2, Config.subscriptionHttpWorkers); + assertEquals(500, Config.subscriptionSocketTimeoutMs); + } + + /** 配置文件缺失不是致命错误，但要落到默认值而不是保留上一次的加载结果。 */ + @Test + void missingFileFallsBackToDefaults(@TempDir Path root) { + Config.subscriptionHttpPort = 12345; + + Config.loadConfig(root.resolve("absent.properties").toString()); + + assertEquals(8889, Config.subscriptionHttpPort, "缺失文件应回到默认值"); + } + + /** + * 启用订阅同步却没有密钥时必须拒绝启动。 + * 否则备机会以空密钥运行，签名校验形同虚设。 + */ + @Test + void enabledSyncWithoutSecretIsRejected(@TempDir Path root) throws Exception { + Path file = Files.writeString(root.resolve("config.properties"), "SubscriptionSyncEnabled=true"); + + IllegalStateException failure = assertThrows(IllegalStateException.class, + () -> Config.loadConfig(file.toString())); + assertTrue(failure.getMessage().contains("SUBSCRIPTION_SYNC_SECRET")); + } +} diff --git a/src/test/java/lion/CustomUtilTest.java b/src/test/java/lion/CustomUtilTest.java new file mode 100644 index 0000000..d59cfbc --- /dev/null +++ b/src/test/java/lion/CustomUtilTest.java @@ -0,0 +1,61 @@ +package lion; + +import org.junit.jupiter.api.Test; + +import java.io.BufferedReader; +import java.io.StringReader; +import java.net.ServerSocket; + +import static org.junit.jupiter.api.Assertions.*; + +/** 下载链路用到的两个工具方法：断点续传请求头解析与空闲端口探测。 */ +class CustomUtilTest { + + /** Range 头必须原样取出「bytes=...」，供 sendFileRange 计算区间。 */ + @Test + void extractsRangeHeader() throws Exception { + BufferedReader reader = new BufferedReader(new StringReader( + "Host: lionwebsite.xyz\r\nUser-Agent: Mozilla/5.0\r\nRange: bytes=1024-2047\r\n\r\n")); + + assertEquals("bytes=1024-2047", CustomUtil.getRequestHeader(reader)); + } + + /** 没有 Range 头（全新下载）时返回 null，调用方据此从头开始传。 */ + @Test + void returnsNullWhenRangeHeaderIsAbsent() throws Exception { + BufferedReader reader = new BufferedReader(new StringReader( + "Host: lionwebsite.xyz\r\nUser-Agent: Mozilla/5.0\r\n\r\n")); + + assertNull(CustomUtil.getRequestHeader(reader)); + } + + /** 从指定端口开始探测，且返回的端口确实可以立刻绑定。 */ + @Test + void returnsBindableIdlePort() throws Exception { + int start = freePort(); + int found = CustomUtil._findIdlePort(start); + + assertTrue(found >= start, "应在起始端口或其之后找到空闲端口"); + try (ServerSocket socket = new ServerSocket(found)) { + assertEquals(found, socket.getLocalPort(), "返回的端口必须可以绑定"); + } + } + + /** 起始端口被占用时不得返回它，必须继续往后找。 */ + @Test + void skipsOccupiedPort() throws Exception { + try (ServerSocket occupied = new ServerSocket(0)) { + int busy = occupied.getLocalPort(); + + int found = CustomUtil._findIdlePort(busy); + + assertTrue(found > busy, "被占用的端口必须跳过"); + } + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/src/test/java/lion/Externel/BackupSubServerTest.java b/src/test/java/lion/Externel/BackupSubServerTest.java new file mode 100644 index 0000000..1038e27 --- /dev/null +++ b/src/test/java/lion/Externel/BackupSubServerTest.java @@ -0,0 +1,264 @@ +package lion.Externel; + +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 lion.Service.SubscriptionSnapshotStore; +import org.junit.jupiter.api.BeforeAll; +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.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.util.Arrays; +import java.util.Base64; +import java.util.HashMap; +import java.util.HexFormat; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.zip.GZIPOutputStream; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 备机订阅分发 HTTP 服务（端口 8889）。 + * + *

这是订阅用户的唯一入口，此前完全没有测试。这里起真实端口，按线路字节校验 + * 状态码、正文与响应头：按公开 Key 命中、两种订阅格式、Range 断点续传、HEAD、 + * 未知 Key、无快照时的 503、健康检查不泄露、方法不被允许以及不可满足的 Range。 + */ +class BackupSubServerTest { + + private static final String SECRET = "backup-sub-test-secret"; + private static final String PUBLIC_KEY = "public-key-abcdef"; + private static final String V2_CONTENT = "v2ray subscription body"; + private static final String CLASH_CONTENT = "clash subscription body"; + + @TempDir + static Path tempDir; + + static int readyPort; + static int emptyPort; + + @BeforeAll + static void startServers() throws Exception { + // 有有效快照的备机：正常分发。 + SubscriptionSnapshotStore ready = store(tempDir.resolve("ready")); + assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, ready.apply(snapshot()).code()); + readyPort = startServer(ready); + + // 从未收到过快照的备机：必须回 503，而不是把请求静默挂起。 + emptyPort = startServer(store(tempDir.resolve("empty"))); + } + + /** 命中公开 Key：v2 与 clash 各自返回对应正文。 */ + @Test + void servesBothSubscriptionFormatsByPublicKey() throws Exception { + Response v2 = get(readyPort, "/sub/v2/" + PUBLIC_KEY); + assertEquals(200, v2.status()); + assertEquals(V2_CONTENT, v2.bodyAsString()); + assertTrue(v2.header("content-type").startsWith("text/plain")); + + Response clash = get(readyPort, "/sub/cat/" + PUBLIC_KEY); + assertEquals(200, clash.status()); + assertEquals(CLASH_CONTENT, clash.bodyAsString()); + assertTrue(clash.header("content-type").startsWith("text/yaml")); + } + + /** Range 请求必须按断点续传返回 206，并给出正确的 Content-Range。 */ + @Test + void supportsPartialContentForResumedDownloads() throws Exception { + Response response = request(readyPort, "GET", "/sub/v2/" + PUBLIC_KEY, "Range: bytes=0-3"); + + assertEquals(206, response.status()); + assertEquals("v2ra", response.bodyAsString()); + assertEquals("bytes 0-3/" + V2_CONTENT.length(), response.header("content-range")); + assertEquals("4", response.header("content-length")); + } + + /** HEAD 必须有正确的 Content-Length，但不回正文。 */ + @Test + void headReturnsHeadersWithoutBody() throws Exception { + Response response = request(readyPort, "HEAD", "/sub/v2/" + PUBLIC_KEY); + + assertEquals(200, response.status()); + assertEquals(String.valueOf(V2_CONTENT.length()), response.header("content-length")); + assertEquals(0, response.body().length, "HEAD 不得返回正文"); + } + + /** 未知 Key 回 404。 */ + @Test + void unknownKeyIsNotFound() throws Exception { + assertEquals(404, get(readyPort, "/sub/v2/unknown-public-key").status()); + } + + /** Key 过短属于非法请求，直接 404 而不是去查表。 */ + @Test + void tooShortKeyIsRejected() throws Exception { + assertEquals(404, get(readyPort, "/sub/v2/abc").status()); + } + + /** 没有有效快照时必须回 503，让客户端重试，而不是假装有内容。 */ + @Test + void returnsServiceUnavailableWhenNoSnapshotExists() throws Exception { + assertEquals(503, get(emptyPort, "/sub/v2/" + PUBLIC_KEY).status()); + } + + /** 健康检查只报快照状态，不得包含 Key 或订阅正文。 */ + @Test + void healthEndpointReportsStateWithoutLeakingKeyOrContent() throws Exception { + Response response = get(readyPort, "/health/subscription"); + + assertEquals(200, response.status()); + assertTrue(response.header("content-type").startsWith("application/json")); + String body = response.bodyAsString(); + assertTrue(body.contains("ready"), "有快照且新鲜时应报告 ready"); + assertFalse(body.contains(PUBLIC_KEY), "健康检查不得泄露公开 Key"); + assertFalse(body.contains(V2_CONTENT), "健康检查不得泄露订阅正文"); + } + + /** 非 GET/HEAD 方法回 405。 */ + @Test + void rejectsUnsupportedMethods() throws Exception { + assertEquals(405, request(readyPort, "POST", "/sub/v2/" + PUBLIC_KEY).status()); + } + + /** 起点越界的 Range 回 416，而不是返回空 200。 */ + @Test + void unsatisfiableRangeIsRejected() throws Exception { + assertEquals(416, request(readyPort, "GET", "/sub/v2/" + PUBLIC_KEY, "Range: bytes=999-1000").status()); + } + + // ---- 测试脚手架 ---- + + private static SubscriptionSnapshotStore store(Path root) { + return new SubscriptionSnapshotStore(root, SECRET, 3600, 1024 * 1024); + } + + private static int startServer(SubscriptionSnapshotStore store) throws IOException { + int port; + try (ServerSocket probe = new ServerSocket(0)) { + port = probe.getLocalPort(); + } + Thread thread = new Thread(new BackupSubServer(store, port, 2), "backup-sub-test-" + port); + thread.setDaemon(true); + thread.start(); + return port; + } + + private static Response get(int port, String path) throws IOException { + return request(port, "GET", path); + } + + private static Response request(int port, String method, String path, String... headers) throws IOException { + try (Socket socket = new Socket("127.0.0.1", port)) { + socket.setSoTimeout(5_000); + StringBuilder request = new StringBuilder() + .append(method).append(' ').append(path).append(" HTTP/1.1\r\n") + .append("Host: 127.0.0.1\r\n"); + for (String header : headers) + request.append(header).append("\r\n"); + request.append("Connection: close\r\n\r\n"); + socket.getOutputStream().write(request.toString().getBytes(StandardCharsets.US_ASCII)); + socket.getOutputStream().flush(); + + byte[] raw = socket.getInputStream().readAllBytes(); + int split = indexOfHeaderEnd(raw); + assertTrue(split > 0, "响应必须含完整的头部结束标记"); + String head = new String(raw, 0, split, StandardCharsets.ISO_8859_1); + byte[] body = Arrays.copyOfRange(raw, split + 4, raw.length); + + String[] lines = head.split("\r\n"); + int status = Integer.parseInt(lines[0].split(" ")[1]); + Map parsed = new HashMap<>(); + for (int i = 1; i < lines.length; i++) { + int colon = lines[i].indexOf(':'); + if (colon > 0) + parsed.put(lines[i].substring(0, colon).trim().toLowerCase(Locale.ROOT), + lines[i].substring(colon + 1).trim()); + } + return new Response(status, parsed, body); + } + } + + private static int indexOfHeaderEnd(byte[] raw) { + for (int i = 0; i + 3 < raw.length; i++) + if (raw[i] == '\r' && raw[i + 1] == '\n' && raw[i + 2] == '\r' && raw[i + 3] == '\n') + return i; + return -1; + } + + private record Response(int status, Map headers, byte[] body) { + String header(String name) { + return headers.get(name); + } + + String bodyAsString() { + return new String(body, StandardCharsets.UTF_8); + } + } + + private static SubscriptionSnapshotMessage snapshot() throws Exception { + ObjectMapper mapper = CustomUtil.objectMapper; + byte[] v2 = V2_CONTENT.getBytes(StandardCharsets.UTF_8); + byte[] clash = CLASH_CONTENT.getBytes(StandardCharsets.UTF_8); + + SubscriptionAccountSnapshot account = new SubscriptionAccountSnapshot(); + account.setAccountId(1); + account.setEnabled(true); + account.setV2ContentBase64(Base64.getEncoder().encodeToString(v2)); + account.setV2Sha256(sha256(v2)); + account.setClashContentBase64(Base64.getEncoder().encodeToString(clash)); + account.setClashSha256(sha256(clash)); + + SubscriptionBindingSnapshot binding = new SubscriptionBindingSnapshot(); + binding.setPublicKeySha256(sha256(PUBLIC_KEY.getBytes(StandardCharsets.UTF_8))); + binding.setAccountId(1); + + SubscriptionSnapshotPayload payload = new SubscriptionSnapshotPayload(); + payload.setSchemaVersion(1); + payload.setAccounts(List.of(account)); + payload.setBindings(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(System.currentTimeMillis()); + message.setPayloadBase64(Base64.getEncoder().encodeToString(compressed)); + message.setPayloadSha256(sha256(compressed)); + String input = "1\n" + message.getRevision() + "\n" + message.getGeneratedAt() + "\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); + } +} diff --git a/src/test/java/lion/MessageCodecTest.java b/src/test/java/lion/MessageCodecTest.java new file mode 100644 index 0000000..94e6428 --- /dev/null +++ b/src/test/java/lion/MessageCodecTest.java @@ -0,0 +1,232 @@ +package lion; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.LengthFieldBasedFrameDecoder; +import lion.Domain.GalleryTask; +import lion.Message.AbstractMessage; +import lion.Message.MessageCodec; +import lion.Message.Main.AvailableCheckMessage; +import lion.Message.Main.DeleteGalleryMessage; +import lion.Message.Main.DownloadPostMessage; +import lion.Message.Main.DownloadStatusMessage; +import lion.Message.Main.IdentityMessage; +import lion.Message.Main.MaintainMessage; +import lion.Message.Main.ResponseMessage; +import lion.Message.Main.SubscriptionSnapshotMessage; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 自定义 TCP 协议 {@code [1 字节类型][4 字节长度][JSON]} 的编解码。 + * + *

这是节点与主站之间唯一的线路格式：类型字节决定反序列化成哪个消息类， + * 长度前缀决定帧边界。两边任何一处不一致都是掉线或错消息，而此前完全无覆盖。 + */ +class MessageCodecTest { + + /** 每种消息类型都要能按自身类型字节往返，且解出的类型正确。 */ + @Test + void roundTripsEveryMessageType() { + assertRoundTrip(new ResponseMessage(7, (byte) 3), AbstractMessage.RESPONSE_MESSAGE); + assertRoundTrip(downloadPost(), AbstractMessage.DOWNLOAD_POST_MESSAGE); + assertRoundTrip(downloadStatus(), AbstractMessage.DOWNLOAD_STATUS_MESSAGE); + assertRoundTrip(deleteGallery(), AbstractMessage.DELETE_GALLERY_MESSAGE); + assertRoundTrip(identity(), AbstractMessage.IDENTITY_MESSAGE); + assertRoundTrip(new MaintainMessage(), AbstractMessage.MAINTAIN_MESSAGE); + assertRoundTrip(new AvailableCheckMessage(), AbstractMessage.AVAILABLE_CHECK_MESSAGE); + assertRoundTrip(snapshot(), AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE); + } + + /** 帧头就是「类型字节 + 大端 4 字节长度」，长度必须等于 JSON 负载的字节数。 */ + @Test + void writesTypeByteAndBigEndianLengthPrefix() { + ByteBuf frame = encode(new ResponseMessage(7, (byte) 3)); + try { + assertEquals(AbstractMessage.RESPONSE_MESSAGE, frame.readByte(), "首字节必须是消息类型"); + int length = frame.readInt(); + int jsonLength = frame.readableBytes(); + assertEquals(jsonLength, length, "长度前缀必须等于 JSON 负载字节数"); + assertTrue(jsonLength > 0); + } finally { + frame.release(); + } + } + + /** 未知类型字节必须被安静丢弃（不产出消息），而不是抛出或产出错误类型的消息。 */ + @Test + void unknownTypeByteIsDroppedWithoutThrowing() { + EmbeddedChannel channel = new EmbeddedChannel(new MessageCodec()); + try { + byte[] payload = "{}".getBytes(StandardCharsets.UTF_8); + ByteBuf frame = Unpooled.buffer(); + frame.writeByte(99); + frame.writeInt(payload.length); + frame.writeBytes(payload); + + channel.writeInbound(frame); + + assertNull(channel.readInbound(), "未知类型不应产出任何消息"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * 生产管线是「先分帧、再解码」：{@link LengthFieldBasedFrameDecoder} 负责等齐整帧， + * 紧跟其后的 {@link MessageCodec} 才假设长度前缀完整。半个帧到达时必须按兵不动、 + * 等剩余字节，而不能抛异常或产出半条消息。 + */ + @Test + void incompleteFrameIsHeldByFrameDecoderUntilComplete() { + EmbeddedChannel channel = new EmbeddedChannel( + new LengthFieldBasedFrameDecoder(100_000_000, 1, 4), new MessageCodec()); + try { + AvailableCheckMessage probe = new AvailableCheckMessage(); + probe.messageId = 11; + + ByteBuf frame = encode(probe); + byte[] bytes = new byte[frame.readableBytes()]; + frame.getBytes(0, bytes); + frame.release(); + + int half = bytes.length / 2; + ByteBuf firstHalf = Unpooled.copiedBuffer(bytes, 0, half); + ByteBuf secondHalf = Unpooled.copiedBuffer(bytes, half, bytes.length - half); + + assertFalse(channel.writeInbound(firstHalf), "半个帧不应产出消息"); + assertNull(channel.readInbound()); + + channel.writeInbound(secondHalf); + AbstractMessage decoded = channel.readInbound(); + assertInstanceOf(AvailableCheckMessage.class, decoded); + assertEquals(11, decoded.messageId); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** 长度前缀来自线路，必须设上限；否则一个报文就能让对端按声明的长度申请内存。 */ + @Test + void rejectsAbsurdDeclaredLength() { + EmbeddedChannel channel = new EmbeddedChannel(new MessageCodec()); + try { + ByteBuf frame = Unpooled.buffer(); + frame.writeByte(AbstractMessage.RESPONSE_MESSAGE); + frame.writeInt(Integer.MAX_VALUE); + + assertThrows(Exception.class, () -> channel.writeInbound(frame), + "超出上限的帧长度必须被拒绝，而不是按声明分配字节数组"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** + * 序列化失败时绝不能写出半帧。 + * + *

原实现先写类型字节、再序列化，失败时只记日志，于是留下「有类型字节、没有长度和 + * JSON」的残帧，对端的分帧器会永远等长度前缀。现在必须先序列化成功再写帧头。 + */ + @Test + void serializationFailureDoesNotEmitHalfFrame() { + EmbeddedChannel channel = new EmbeddedChannel(new MessageCodec()); + try { + assertThrows(Exception.class, () -> channel.writeOutbound(new UnserializableMessage()), + "序列化失败必须抛错，而不是安静地留下半帧"); + assertNull(channel.readOutbound(), "失败时不得产出任何字节"); + } finally { + channel.finishAndReleaseAll(); + } + } + + /** 故意让 Jackson 在序列化属性时抛异常。 */ + static class UnserializableMessage extends AbstractMessage { + UnserializableMessage() { + messageType = AbstractMessage.RESPONSE_MESSAGE; + } + + @SuppressWarnings("unused") + public String getExplode() { + throw new IllegalStateException("boom"); + } + } + + private static void assertRoundTrip(AbstractMessage message, byte expectedType) { + ByteBuf frame = encode(message); + assertEquals(expectedType, frame.getByte(0), "线路上的类型字节必须与消息类匹配"); + + EmbeddedChannel channel = new EmbeddedChannel( + new LengthFieldBasedFrameDecoder(100_000_000, 1, 4), new MessageCodec()); + try { + channel.writeInbound(frame); + AbstractMessage decoded = channel.readInbound(); + assertNotNull(decoded, "整帧必须被解码出来"); + assertEquals(message.getClass(), decoded.getClass(), "解码类型必须与编码类型一致"); + assertEquals(expectedType, decoded.messageType); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static ByteBuf encode(AbstractMessage message) { + EmbeddedChannel channel = new EmbeddedChannel(new MessageCodec()); + try { + channel.writeOutbound(message); + ByteBuf out = channel.readOutbound(); + assertNotNull(out, "编码后必须产出字节"); + return out; + } finally { + channel.finishAndReleaseAll(); + } + } + + private static DownloadPostMessage downloadPost() { + GalleryTask task = new GalleryTask(); + task.setGid(4242); + task.setName("sample [4242]"); + task.setStatus(GalleryTask.DOWNLOADING); + DownloadPostMessage message = new DownloadPostMessage(); + message.messageId = 1; + message.setGalleryTask(task); + return message; + } + + private static DownloadStatusMessage downloadStatus() { + GalleryTask task = new GalleryTask(); + task.setGid(4242); + task.setName("sample [4242]"); + task.setStatus(GalleryTask.COMPRESS_COMPLETE); + DownloadStatusMessage message = new DownloadStatusMessage(); + message.setGalleryTasks(new GalleryTask[]{task}); + return message; + } + + private static DeleteGalleryMessage deleteGallery() { + DeleteGalleryMessage message = new DeleteGalleryMessage(); + message.messageId = 5; + message.setGalleryName("sample [4242]"); + return message; + } + + private static IdentityMessage identity() { + IdentityMessage message = new IdentityMessage(); + message.setIdentity("lionwebsite"); + return message; + } + + private static SubscriptionSnapshotMessage snapshot() { + SubscriptionSnapshotMessage message = new SubscriptionSnapshotMessage(); + message.setSchemaVersion(1); + message.setRevision("a".repeat(64)); + message.setGeneratedAt(1_700_000_000_000L); + message.setPayloadBase64("ZmFrZQ=="); + message.setPayloadSha256("b".repeat(64)); + message.setSignature("c".repeat(64)); + return message; + } +} diff --git a/src/test/java/lion/MultiThreadedHTTPServerTest.java b/src/test/java/lion/MultiThreadedHTTPServerTest.java index 5faa3c5..0bf2772 100644 --- a/src/test/java/lion/MultiThreadedHTTPServerTest.java +++ b/src/test/java/lion/MultiThreadedHTTPServerTest.java @@ -1,8 +1,16 @@ package lion; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; @@ -58,4 +66,123 @@ class MultiThreadedHTTPServerTest { void returnsNullForNullRequestLine() { assertNull(MultiThreadedHTTPServer.parseRequestLine(null)); } + + // ---- 请求处理（真实 socket） ---- + + /** + * 畸形请求行（没有空格）不得打死处理线程，更不得泄漏连接。 + * + *

原实现直接取 {@code requestParts[1]}，空行会抛 ArrayIndexOutOfBoundsException； + * 它不是 IOException，会被线程池静默吞掉，socket 也关不掉。现在应回 400 并关闭连接。 + */ + @Test + void malformedRequestLineIsRejectedWithoutCrashing() throws Exception { + try (ServerSocket server = new ServerSocket(0); Socket client = new Socket("127.0.0.1", server.getLocalPort())) { + client.setSoTimeout(5_000); + Thread handler = startHandler(server); + + client.getOutputStream().write("GARBAGE\r\n\r\n".getBytes(StandardCharsets.US_ASCII)); + client.getOutputStream().flush(); + + String head = readHead(client); + assertTrue(head.startsWith("HTTP/1.1 400"), "畸形请求行应回 400，实际：" + head.lines().findFirst().orElse("")); + handler.join(5_000); + assertFalse(handler.isAlive(), "处理器必须结束并关闭连接，不得挂住线程"); + } + } + + /** 只有空行的连接应被安静关闭，不产生响应也不抛异常。 */ + @Test + void blankRequestLineClosesConnectionQuietly() throws Exception { + try (ServerSocket server = new ServerSocket(0); Socket client = new Socket("127.0.0.1", server.getLocalPort())) { + client.setSoTimeout(5_000); + AtomicReference failure = new AtomicReference<>(); + Thread handler = startHandler(server, failure); + + client.getOutputStream().write("\r\n".getBytes(StandardCharsets.US_ASCII)); + client.getOutputStream().flush(); + + assertEquals(-1, client.getInputStream().read(), "应直接关闭连接而不回正文"); + handler.join(5_000); + assertFalse(handler.isAlive(), "处理器必须结束"); + assertNull(failure.get(), "空行不得抛异常： " + failure.get()); + } + } + + /** + * 请求的 Range 超过文件末端时必须夹到文件末尾。 + * + *

原实现按未夹取的区间写 Content-Length，客户端拿到一个远大于实际发送量的长度， + * 会一直等下去。现在 Content-Range / Content-Length / 正文都必须只覆盖到文件末尾。 + */ + @Test + void oversizedRangeIsClampedToFileEnd(@TempDir Path root) throws Exception { + byte[] payload = "0123456789".getBytes(StandardCharsets.UTF_8); + Path archive = root.resolve("archive.zip"); + Files.write(archive, payload); + + String head = fetchRange(root, archive, "bytes=4-999999"); + + assertTrue(head.contains("Content-Range: bytes 4-9/10"), "Content-Range 必须夹到文件末尾：" + head); + assertTrue(head.contains("Content-Length: 6"), "Content-Length 必须与实际发送量一致：" + head); + } + + /** Range 起点超过文件长度时无可发送内容，应回 416，而不是假装有长度。 */ + @Test + void rangeStartingBeyondFileIsNotSatisfiable(@TempDir Path root) throws Exception { + Path archive = root.resolve("archive.zip"); + Files.write(archive, "0123456789".getBytes(StandardCharsets.UTF_8)); + + String head = fetchRange(root, archive, "bytes=999-1000"); + + assertTrue(head.startsWith("HTTP/1.1 416"), "越界起点应回 416：" + head.lines().findFirst().orElse("")); + } + + // ---- 测试脚手架 ---- + + /** 在守护线程里接受一个连接并交给真实的请求处理器。 */ + private static Thread startHandler(ServerSocket server) { + return startHandler(server, new AtomicReference<>()); + } + + private static Thread startHandler(ServerSocket server, AtomicReference failure) { + Thread handler = new Thread(() -> { + try { + MultiThreadedHTTPServer.handleClientRequest(server.accept()); + } catch (Throwable t) { + failure.set(t); + } + }); + handler.setDaemon(true); + handler.start(); + return handler; + } + + /** 走 AuthCode=alone 路径请求指定文件，返回响应头。 */ + private static String fetchRange(Path root, Path archive, String range) throws IOException { + try (ServerSocket server = new ServerSocket(0); Socket client = new Socket("127.0.0.1", server.getLocalPort())) { + client.setSoTimeout(5_000); + Thread handler = startHandler(server); + + String request = "GET " + archive + "?AuthCode=alone HTTP/1.1\r\n" + + "Host: 127.0.0.1\r\nRange: " + range + "\r\n\r\n"; + client.getOutputStream().write(request.getBytes(StandardCharsets.US_ASCII)); + client.getOutputStream().flush(); + + return readHead(client); + } + } + + private static String readHead(Socket socket) throws IOException { + java.io.ByteArrayOutputStream buffer = new java.io.ByteArrayOutputStream(); + int b; + while ((b = socket.getInputStream().read()) != -1) { + buffer.write(b); + byte[] bytes = buffer.toByteArray(); + if (bytes.length >= 4 && bytes[bytes.length - 4] == '\r' && bytes[bytes.length - 3] == '\n' + && bytes[bytes.length - 2] == '\r' && bytes[bytes.length - 1] == '\n') + break; + } + return buffer.toString(StandardCharsets.ISO_8859_1); + } } diff --git a/src/test/java/lion/Service/DeleteServiceTest.java b/src/test/java/lion/Service/DeleteServiceTest.java new file mode 100644 index 0000000..3b33bc4 --- /dev/null +++ b/src/test/java/lion/Service/DeleteServiceTest.java @@ -0,0 +1,82 @@ +package lion.Service; + +import lion.ErrorCode.ErrorCode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.jupiter.api.Assertions.*; + +/** 画廊目录删除：成功归零，失败按 ErrorCode 区分「路径不存在」与其它 IO 错误。 */ +class DeleteServiceTest { + + @Test + void deletesExistingDirectoryAndReturnsSuccess(@TempDir Path root) throws Exception { + Path gallery = Files.createDirectories(root.resolve("sample [123]")); + Files.writeString(gallery.resolve("1.jpg"), "image bytes"); + + assertEquals(0, DeleteService.deleteAll(gallery.toString())); + assertFalse(Files.exists(gallery), "目录应被整棵删除"); + } + + /** 路径不存在时必须回 FILE_NOT_FOUND，让主站能区分「本来就没有」。 */ + @Test + void missingPathReportsFileNotFound(@TempDir Path root) { + assertEquals(ErrorCode.FILE_NOT_FOUND, DeleteService.deleteAll(root.resolve("absent").toString())); + } + + /** 目标是普通文件（而非目录）时同样按 FILE_NOT_FOUND 处理，不得误删文件。 */ + @Test + void plainFileIsNotTreatedAsDirectory(@TempDir Path root) throws Exception { + Path file = root.resolve("not-a-directory.txt"); + Files.writeString(file, "keep me"); + + assertEquals(ErrorCode.FILE_NOT_FOUND, DeleteService.deleteAll(file.toString())); + assertTrue(Files.exists(file), "普通文件不应被删除"); + } + + /** 正常的子目录名应在根目录内被删除。 */ + @Test + void deleteWithinRemovesChildOfRoot(@TempDir Path root) throws Exception { + Path gallery = Files.createDirectories(root.resolve("sample [123]")); + + assertEquals(0, DeleteService.deleteWithin(root.toString(), "sample [123]")); + assertFalse(Files.exists(gallery)); + } + + /** + * 画廊名来自主站下发的消息，带 {@code ../} 时必须被拒绝。 + * 否则路径会被规范化到根目录之外，删掉节点上的任意目录。 + */ + @Test + void deleteWithinRejectsPathTraversal(@TempDir Path root) throws Exception { + Path outside = Files.createDirectories(root.resolve("important")); + Path galleryRoot = Files.createDirectories(root.resolve("gallery")); + + byte result = DeleteService.deleteWithin(galleryRoot.toString(), "../important"); + + assertEquals(ErrorCode.FILE_NOT_FOUND, result, "越界路径必须按找不到处理"); + assertTrue(Files.exists(outside), "根目录之外的内容绝不能被删除"); + } + + /** 绝对路径同样不得越过根目录。 */ + @Test + void deleteWithinRejectsAbsolutePath(@TempDir Path root) throws Exception { + Path outside = Files.createDirectories(root.resolve("important")); + Path galleryRoot = Files.createDirectories(root.resolve("gallery")); + + assertEquals(ErrorCode.FILE_NOT_FOUND, DeleteService.deleteWithin(galleryRoot.toString(), outside.toString())); + assertTrue(Files.exists(outside)); + } + + /** 空名会导致规范化结果等于根目录本身，必须拒绝。 */ + @Test + void deleteWithinRejectsEmptyName(@TempDir Path root) throws Exception { + Path galleryRoot = Files.createDirectories(root.resolve("gallery")); + + assertEquals(ErrorCode.FILE_NOT_FOUND, DeleteService.deleteWithin(galleryRoot.toString(), " ")); + assertTrue(Files.exists(galleryRoot), "根目录本身不得被删除"); + } +} diff --git a/src/test/java/lion/Service/DownloadCheckServiceGidMatchingTest.java b/src/test/java/lion/Service/DownloadCheckServiceGidMatchingTest.java new file mode 100644 index 0000000..e463921 --- /dev/null +++ b/src/test/java/lion/Service/DownloadCheckServiceGidMatchingTest.java @@ -0,0 +1,113 @@ +package lion.Service; + +import lion.Domain.GalleryTask; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.ConcurrentHashMap; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 入队时按 gid 定位磁盘内容。 + * + *

gid 稳定、目录名不稳定（下载器会追加分辨率等后缀），因此匹配不能只看名字相等， + * 而要按 {@code [gid]} / {@code [gid-...]} 标记判断；同时绝不能把 {@code [1234]} + * 误判成 gid 123，否则会把下载链接指向错误的画廊。 + */ +class DownloadCheckServiceGidMatchingTest { + + /** 目录名带后缀时也要能按 gid 命中已有归档，并判为压缩完成。 */ + @Test + void matchesStoredArchiveWhenDirectoryNameHasSuffix(@TempDir Path root) throws Exception { + DownloadCheckService service = service(root); + Path directory = Files.createDirectories( + Path.of(service.storagePath, "sample [123-456]")); + writeValidZip(directory.resolve("sample [123-456].zip")); + + GalleryTask result = service.addToQueue(task(123)); + + assertTrue(result.is_compress_complete(), "gid 命中已有归档应判为已完成"); + assertEquals("sample [123-456]", result.getName(), "名称应取实际目录名"); + } + + /** 尚未出现 galleryinfo.txt：判为下载中，并统计已下载页数。 */ + @Test + void marksTaskDownloadingWhileGalleryInfoMissing(@TempDir Path root) throws Exception { + DownloadCheckService service = service(root); + Path directory = Files.createDirectories(Path.of(service.downloadPath, "sample [123-456]")); + Files.writeString(directory.resolve("1.jpg"), "x"); + Files.writeString(directory.resolve("2.jpg"), "y"); + + GalleryTask result = service.addToQueue(task(123)); + + assertEquals(GalleryTask.DOWNLOADING, result.getStatus()); + assertEquals(2, result.getProceeding(), "应统计除 galleryinfo.txt 外的页数"); + assertEquals("sample [123-456]", result.getName()); + } + + /** 出现 galleryinfo.txt：判为下载完成，并记录源目录用于后续压缩。 */ + @Test + void marksTaskCompleteWhenGalleryInfoPresent(@TempDir Path root) throws Exception { + DownloadCheckService service = service(root); + Path directory = Files.createDirectories(Path.of(service.downloadPath, "sample [123]")); + Files.writeString(directory.resolve("1.jpg"), "x"); + Files.writeString(directory.resolve("galleryinfo.txt"), "meta"); + + GalleryTask result = service.addToQueue(task(123)); + + assertEquals(GalleryTask.DOWNLOAD_COMPLETE, result.getStatus()); + assertEquals(directory.toString(), result.getPath()); + } + + /** gid 前缀相同但实际不同的目录不得被匹配。 */ + @Test + void doesNotMatchLongerGidSharingTheSamePrefix(@TempDir Path root) throws Exception { + DownloadCheckService service = service(root); + Files.createDirectories(Path.of(service.downloadPath, "sample [1234]")); + + GalleryTask result = service.addToQueue(task(123)); + + assertEquals(0, result.getStatus(), "[1234] 不得被当成 gid 123 的目录"); + assertNull(result.getName(), "未命中时不应改写任务名称"); + } + + /** 归档校验：空文件与不存在的路径都不算有效归档。 */ + @Test + void rejectsEmptyAndMissingArchives(@TempDir Path root) throws Exception { + Path empty = Files.createFile(root.resolve("empty.zip")); + assertFalse(DownloadCheckService.isValidArchive(empty.toFile())); + assertFalse(DownloadCheckService.isValidArchive(root.resolve("missing.zip").toFile())); + + Path valid = root.resolve("valid.zip"); + writeValidZip(valid); + assertTrue(DownloadCheckService.isValidArchive(valid.toFile())); + } + + private static DownloadCheckService service(Path root) throws Exception { + DownloadCheckService service = new DownloadCheckService(new ConcurrentHashMap<>(), false); + service.downloadPath = Files.createDirectories(root.resolve("download")).toString(); + service.storagePath = Files.createDirectories(root.resolve("gallery")).toString(); + return service; + } + + private static GalleryTask task(int gid) { + GalleryTask task = new GalleryTask(); + task.setGid(gid); + return task; + } + + private static void writeValidZip(Path zip) throws Exception { + Files.createDirectories(zip.getParent()); + try (ZipOutputStream out = new ZipOutputStream(Files.newOutputStream(zip))) { + out.putNextEntry(new ZipEntry("1.jpg")); + out.write("image bytes".getBytes(StandardCharsets.UTF_8)); + out.closeEntry(); + } + } +} diff --git a/src/test/java/lion/StorageNodeMessageHandlingTest.java b/src/test/java/lion/StorageNodeMessageHandlingTest.java new file mode 100644 index 0000000..3f0531f --- /dev/null +++ b/src/test/java/lion/StorageNodeMessageHandlingTest.java @@ -0,0 +1,297 @@ +package lion; + +import io.netty.channel.embedded.EmbeddedChannel; +import lion.Config.Config; +import lion.Domain.GalleryTask; +import lion.ErrorCode.ErrorCode; +import lion.Message.Main.AvailableCheckMessage; +import lion.Message.Main.DeleteGalleryMessage; +import lion.Message.Main.DownloadPostMessage; +import lion.Message.Main.DownloadStatusMessage; +import lion.Message.Main.IdentityMessage; +import lion.Message.Main.ResponseMessage; +import lion.Message.Main.SubscriptionSnapshotMessage; +import lion.Service.SubscriptionSnapshotStore; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 节点消息处理与状态上报循环。 + * + *

{@link storageNode.MyChannelInboundHandlerAdapter} 与 {@link storageNode#mainThread()} + * 约 250 行，承载任务下发、删除、探活、订阅快照分发与队列重放，此前只有静态方法 + * {@code isPrimaryEvidence} 被间接覆盖。这里用 {@link EmbeddedChannel} 把入站消息直接 + * 喂给真实处理器，断言回包与队列副作用，同时锁定「无可用通道时不得丢弃任务」的语义。 + */ +class StorageNodeMessageHandlingTest { + + @TempDir + static Path tempDir; + + static storageNode node; + + @BeforeAll + static void startNode() throws Exception { + Path storageRoot = Files.createDirectories(tempDir.resolve("gallery")); + Path subscriptionRoot = Files.createDirectories(tempDir.resolve("sub")); + + storageNode.storagePath = storageRoot + "/"; + SubscriptionSnapshotStore store = + new SubscriptionSnapshotStore(subscriptionRoot, "test-secret", 3600, 1024 * 1024); + node = new storageNode(store); + // 订阅同步默认关闭：快照消息必须被拒绝，也不触发异步落盘。 + Config.subscriptionSyncEnabled = false; + } + + @AfterAll + static void stopNode() { + node.subscriptionApplyExecutor.shutdownNow(); + node.rekickExecutor.shutdownNow(); + } + + @BeforeEach + void resetState() { + node.queue.clear(); + node.tempQueue.clear(); + } + + /** + * 任务下发：先按 gid 入队并立即回传当前状态，再回执响应。 + * + *

回传顺序不能反——主站的单任务重试接口依赖「先收到状态、再收到响应」。 + */ + @Test + void downloadPostEnqueuesTaskAndReportsStatusBeforeResponse() { + EmbeddedChannel channel = nodeChannel(); + try { + DownloadPostMessage post = new DownloadPostMessage(); + post.setGalleryTask(task(90001)); + post.messageId = 7; + + channel.writeInbound(post); + + DownloadStatusMessage status = channel.readOutbound(); + assertNotNull(status, "应先把当前任务状态回传给主站"); + assertEquals(1, status.getGalleryTasks().length); + assertEquals(90001, status.getGalleryTasks()[0].getGid()); + + ResponseMessage response = channel.readOutbound(); + assertNotNull(response, "随后应回执响应"); + assertEquals(7, response.messageId, "响应必须带上下发方的 messageId"); + assertEquals(0, response.getResult()); + + assertTrue(node.queue.containsKey(90001), "任务应已进入待处理队列"); + assertTrue(node.primaryChannel.usable(), "主站消息应认领通道引用"); + } finally { + close(channel); + } + } + + /** 删除任务：既清理待上报队列，也删掉磁盘目录，并回执成功。 */ + @Test + void deleteGalleryRemovesQueuedTaskAndDirectory() throws Exception { + EmbeddedChannel channel = nodeChannel(); + try { + Path directory = Files.createDirectories(Path.of(storageNode.storagePath, "sample [90002]")); + node.queue.put(90002, task(90002)); + + DeleteGalleryMessage delete = new DeleteGalleryMessage(); + delete.setGalleryName("sample [90002]"); + delete.messageId = 3; + channel.writeInbound(delete); + + ResponseMessage response = channel.readOutbound(); + assertNotNull(response); + assertEquals(3, response.messageId); + assertEquals(0, response.getResult(), "目录删除成功应回 0"); + + assertFalse(node.queue.containsKey(90002), "已删除的任务不得继续上报状态"); + assertFalse(Files.exists(directory), "磁盘目录应被删除"); + } finally { + close(channel); + } + } + + /** 删除不存在的目录时仍要回包，并如实返回 FILE_NOT_FOUND。 */ + @Test + void deleteGalleryMissingDirectoryReportsFileNotFound() { + EmbeddedChannel channel = nodeChannel(); + try { + DeleteGalleryMessage delete = new DeleteGalleryMessage(); + delete.setGalleryName("never-existed [90003]"); + delete.messageId = 4; + + channel.writeInbound(delete); + + ResponseMessage response = channel.readOutbound(); + assertNotNull(response, "无论删除成败都必须回执，否则主站会一直等待"); + assertEquals(4, response.messageId); + assertEquals(ErrorCode.FILE_NOT_FOUND, response.getResult()); + } finally { + close(channel); + } + } + + /** 探活：必须原样带回 messageId，供主站配对。 */ + @Test + void availableCheckEchoesMessageId() { + EmbeddedChannel channel = nodeChannel(); + try { + AvailableCheckMessage probe = new AvailableCheckMessage(); + probe.messageId = 42; + + channel.writeInbound(probe); + + ResponseMessage response = channel.readOutbound(); + assertNotNull(response); + assertEquals(42, response.messageId); + assertEquals(0, response.getResult()); + } finally { + close(channel); + } + } + + /** 主站身份消息认领通道引用；备机身份消息不得认领，否则状态会上报到错误连接。 */ + @Test + void onlyPrimaryIdentityClaimsChannelReference() { + EmbeddedChannel primary = nodeChannel(); + EmbeddedChannel sidecar = nodeChannel(); + try { + sidecar.writeInbound(identity("lionwebsiteside")); + assertFalse(node.primaryChannel.usable(), "备机身份不得占用主站引用"); + assertSame(sidecar, node.node, "备机通道应登记为 node"); + + primary.writeInbound(identity("lionwebsite")); + assertTrue(node.primaryChannel.usable(), "主站身份应认领引用"); + assertSame(primary, node.primaryChannel.current()); + } finally { + close(primary); + close(sidecar); + } + } + + /** 未开启订阅同步时，快照消息必须被拒绝。 */ + @Test + void subscriptionSnapshotRejectedWhenSyncDisabled() { + EmbeddedChannel channel = nodeChannel(); + try { + SubscriptionSnapshotMessage snapshot = new SubscriptionSnapshotMessage(); + snapshot.messageId = 9; + snapshot.setSchemaVersion(1); + snapshot.setRevision("a".repeat(64)); + + channel.writeInbound(snapshot); + + ResponseMessage response = channel.readOutbound(); + assertNotNull(response); + assertEquals(9, response.messageId); + assertEquals(SubscriptionSnapshotStore.APPLY_INVALID, response.getResult()); + } finally { + close(channel); + } + } + + /** 有待上报任务且通道可用时，mainThread 应上报并清空已完成的压缩任务。 */ + @Test + void mainThreadReportsCompletedTaskAndDrainsQueue() { + EmbeddedChannel channel = nodeChannel(); + try { + channel.writeInbound(identity("lionwebsite")); + drainOutbound(channel); + + GalleryTask done = task(90004); + done.setStatus(GalleryTask.COMPRESS_COMPLETE); + node.queue.put(90004, done); + + node.mainThread(); + + DownloadStatusMessage status = readStatusMessage(channel); + assertNotNull(status, "通道可用时必须上报任务状态"); + assertEquals(1, status.getGalleryTasks().length); + assertEquals(90004, status.getGalleryTasks()[0].getGid()); + assertFalse(node.queue.containsKey(90004), "压缩完成的任务上报后应出队"); + } finally { + close(channel); + } + } + + /** + * 无可用通道时不得上报，也绝不能丢弃任务——队列要保留到主站重连后重放。 + * 这是「状态永久卡在已提交」的另一半成因。 + */ + @Test + void mainThreadKeepsTasksWhenNoChannelIsUsable() { + assertFalse(node.primaryChannel.usable(), "前置条件：本用例开始时不应有已认证通道"); + + GalleryTask done = task(90005); + done.setStatus(GalleryTask.COMPRESS_COMPLETE); + node.queue.put(90005, done); + + node.mainThread(); + + assertTrue(node.queue.containsKey(90005), "无通道时任务必须保留在队列中等待重放"); + assertNull(node.primaryChannel.current()); + } + + /** 队列为空时不得发送任务状态消息（心跳不算）。 */ + @Test + void mainThreadSendsNoStatusWhenQueueIsEmpty() { + EmbeddedChannel channel = nodeChannel(); + try { + channel.writeInbound(identity("lionwebsite")); + drainOutbound(channel); + + node.mainThread(); + + assertNull(readStatusMessage(channel), "没有任务可报时不应发送状态消息"); + } finally { + close(channel); + } + } + + private static EmbeddedChannel nodeChannel() { + return new EmbeddedChannel(node.new MyChannelInboundHandlerAdapter(node.tempQueue)); + } + + private static GalleryTask task(int gid) { + GalleryTask task = new GalleryTask(); + task.setGid(gid); + task.setName("sample [" + gid + "]"); + return task; + } + + private static IdentityMessage identity(String value) { + IdentityMessage message = new IdentityMessage(); + message.setIdentity(value); + return message; + } + + /** 读出并丢弃全部出站消息，用于清场。 */ + private static int drainOutbound(EmbeddedChannel channel) { + int count = 0; + while (channel.readOutbound() != null) + count++; + return count; + } + + /** 找出第一条任务状态消息；容忍调度线程可能插入的心跳等消息。 */ + private static DownloadStatusMessage readStatusMessage(EmbeddedChannel channel) { + Object outbound; + while ((outbound = channel.readOutbound()) != null) + if (outbound instanceof DownloadStatusMessage status) + return status; + return null; + } + + private static void close(EmbeddedChannel channel) { + channel.finishAndReleaseAll(); + } +}