加固下载服务与删除路径,补测试

- 8888 下载服务:加读超时、接受循环单次失败不退出;畸形请求行回 400 且不泄漏连接;
  Range 越界按文件末端夹取或回 416,非法 Range 不再中断处理
- 删除任务改为按根目录规范化校验,拦截 ../ 与绝对路径越界删除
- MessageCodec:先序列化再写帧头(不再产生半帧),并限制单帧长度上限
- 节点绑定端口等待结果,失败即抛出而非静默不监听
- 配置加载支持指定路径并对缺文件记录告警;完成通知对消息做 URL 编码并加超时
- 测试 28 → 76:新增协议编解码、节点消息处理与上报循环、备机订阅分发、
  gid 匹配、删除边界、配置加载等用例
This commit is contained in:
us9929
2026-09-23 01:04:03 +08:00
parent 77f817d402
commit 80b82121a1
15 changed files with 1435 additions and 58 deletions
+6 -2
View File
@@ -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 反射配置
+39 -12
View File
@@ -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);
}
/**
* 从指定文件加载配置。
*
* <p>拆出带路径的版本是为了能直接对「默认值填充」和「启用同步却没配密钥就拒绝启动」
* 写断言,不必依赖生产路径存在。文件缺失不是致命错误(保留内置默认值),
* 但会记录明确日志,避免以错误配置静默运行。
*/
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) {
+21 -3
View File
@@ -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");
+15 -6
View File
@@ -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<AbstractMessage> {
/** 单帧负载上限,与上游 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<AbstractMessage> {
@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<Object> 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);
+63 -32
View File
@@ -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<String, String> 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());
}
}
@@ -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;
}
/**
* 删除某个根目录下的子目录,且保证不越界。
*
* <p>画廊名来自主站下发的消息,直接拼接根路径会让 {@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());
}
}
+13 -3
View File
@@ -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<NioSocketChannel>() {
@@ -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);
}
+82
View File
@@ -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.*;
/**
* 配置加载。
*
* <p>生产配置来自 {@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"));
}
}
+61
View File
@@ -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();
}
}
}
@@ -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)。
*
* <p>这是订阅用户的唯一入口,此前完全没有测试。这里起真实端口,按线路字节校验
* 状态码、正文与响应头:按公开 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<String, String> 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<String, String> 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);
}
}
+232
View File
@@ -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]} 的编解码。
*
* <p>这是节点与主站之间唯一的线路格式:类型字节决定反序列化成哪个消息类,
* 长度前缀决定帧边界。两边任何一处不一致都是掉线或错消息,而此前完全无覆盖。
*/
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();
}
}
/**
* 序列化失败时绝不能写出半帧。
*
* <p>原实现先写类型字节、再序列化,失败时只记日志,于是留下「有类型字节、没有长度和
* 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;
}
}
@@ -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) ----
/**
* 畸形请求行(没有空格)不得打死处理线程,更不得泄漏连接。
*
* <p>原实现直接取 {@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<Throwable> 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 超过文件末端时必须夹到文件末尾。
*
* <p>原实现按未夹取的区间写 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<Throwable> 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);
}
}
@@ -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), "根目录本身不得被删除");
}
}
@@ -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 定位磁盘内容。
*
* <p>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();
}
}
}
@@ -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.*;
/**
* 节点消息处理与状态上报循环。
*
* <p>{@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 入队并立即回传当前状态,再回执响应。
*
* <p>回传顺序不能反——主站的单任务重试接口依赖「先收到状态、再收到响应」。
*/
@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();
}
}