diff --git a/SUBSCRIPTION_GATEWAY.md b/SUBSCRIPTION_GATEWAY.md new file mode 100644 index 0000000..34d3dfd --- /dev/null +++ b/SUBSCRIPTION_GATEWAY.md @@ -0,0 +1,45 @@ +# storageNode 内置订阅网关 + +8889 的 BackupSubServer 直接承载网关；无需独立进程或端口。空的 +`SubscriptionPrimaryUrl` 保留旧快照模式；生产设 `https://lionwebsite.xyz`， +`SubscriptionEventDir` 默认 `/root/gallery/storageNode/subscription-events`， +共享 `SubscriptionSyncSecret`，启用网关且无密钥会拒绝启动。 + +Nginx `/sub/` 始终代理本机8889，并覆盖 `X-Real-IP $remote_addr`，保留原始 +User-Agent/Range。只有 loopback 对端可以提供 X-Real-IP；公网直连8889 +忽略所有代理来源 IP，X-Forwarded-For 始终不使用。 + +每个有效 v2/cat GET/HEAD 优先请求固定主站 origin，连接1秒、整次含完整正文3秒； +不传外部认证头、不跟随重定向，正文最大16MiB。网络错误、超时、5xx及3xx转备用快照； +400/403/404等其他状态透传，不使用旧缓存。GET/HEAD、Range和内容类型/范围头保留。 +失败后熔断10秒，期间只读本地，到期只允许一个探测，成功后恢复逐请求访问主站。 + +每次请求建立UUID事件，字段为eventId/keyHash/client/clientIp/userAgent/occurredAt。 +只存公开key的SHA256，不保存公开key、正文或密钥。UA保留合法Latin-1原始前缀，去除控制字符、非Latin-1替换为?，最大2048字符且UTF8<=2048bytes， +并在实际base64url事件头超过4096字符时继续截断；发主站UA与事件一致。 +GET和HEAD均携带签名事件头；HEAD不会记日志。 + +签名请求头为 X-Subscription-Event（base64url无padding紧凑JSON）、 +X-Subscription-Timestamp（epoch秒）、X-Subscription-Signature（HMAC-SHA256 hex）。 +签名输入 `v1\n\n`，UTF8。 +主站已记但响应超时，也用原eventId回传，由主站receipt去重。 + +备用快照200/206 GET必须先写事件文件并fsync/原子rename/目录fsync才返回； +落盘失败或队列超限返回503，HEAD/404/416/503不写成功事件。 +目录700，文件600。上限100000条/256MiB，保留16MiB磁盘余量；达到限制拒绝新成功更新， +不会静默丢弃旧记录。启动恢复所有json，元数据损坏会拒绝网关初始化，需先保留文件并排查。 + +后台单线程每10秒最多回传1批，100条且原始body<=256KiB，POST +`/internal/subscription-events`，body `{"events":[...]}`。签名使用同一规则对原始body计算， +不含事件header。只清理本批 `{"acknowledged":[""]}` 明确确认的事件， +并同步目录。网络/5xx/非法JSON/未确认/未知ID均保留原ID重试。上传不持有熔断器锁。 +进程关闭会释放HTTP client和后台线程；Main shutdown hook管理完整服务生命周期。 + +`/health/subscription` 保持现有快照字段；`/health/subscription-gateway` +另报pendingCount/oldestAgeMillis/circuit/lastUploadAt/lastUploadState，无用户信息。 + +验证必须在部署手册指定的构建VM运行，不能在工作机执行Java编译/测试： +`mvn -B test`。`SubscriptionGatewayTest` 使用真实本地HTTP主站和节点入口覆盖签名、 +主站优先、fallback/持久化、ack/重启/同ID、HEAD、404、Range、timeout、熔断/单探测、 +后台上传隔离、来源IP、路径编码、磁盘错误、批次字节上限、redirect和不完整/超大正文。 +新增持久化协议全用Jackson树节点，不新增反射DTO。 diff --git a/src/main/java/lion/Config/Config.java b/src/main/java/lion/Config/Config.java index 70b204f..2583777 100644 --- a/src/main/java/lion/Config/Config.java +++ b/src/main/java/lion/Config/Config.java @@ -21,6 +21,9 @@ public class Config { public static int subscriptionHttpPort; public static int subscriptionHttpWorkers; public static int subscriptionSocketTimeoutMs; + /** 空值保留快照直读模式；设置后启用进程内主站优先网关。 */ + public static String subscriptionPrimaryUrl = ""; + public static String subscriptionEventDir = "/root/gallery/storageNode/subscription-events"; /** 下载服务的管理员口令；为空表示 8888 的管理员直取文件能力关闭。 */ public static String adminDownloadCode = ""; @@ -76,6 +79,10 @@ public class Config { subscriptionHttpPort = Integer.parseInt(value(prop, "SubscriptionHttpPort", "8889")); subscriptionHttpWorkers = Integer.parseInt(value(prop, "SubscriptionHttpWorkers", "4")); subscriptionSocketTimeoutMs = Integer.parseInt(value(prop, "SubscriptionSocketTimeoutMs", "10000")); + subscriptionPrimaryUrl = value(prop, "SubscriptionPrimaryUrl", ""); + subscriptionEventDir = value(prop, "SubscriptionEventDir", "/root/gallery/storageNode/subscription-events"); + if (!subscriptionPrimaryUrl.isBlank() && subscriptionSyncSecret.isBlank()) + throw new IllegalStateException("启用订阅网关时必须配置 SUBSCRIPTION_SYNC_SECRET"); // 口令不再写死在源码里：优先取环境变量，其次取配置文件，都为空即关闭该能力。 adminDownloadCode = System.getenv().getOrDefault("STORAGE_DOWNLOAD_ADMIN_CODE", value(prop, "AdminDownloadCode", "")); diff --git a/src/main/java/lion/Externel/BackupSubServer.java b/src/main/java/lion/Externel/BackupSubServer.java index 91bcda8..5edac61 100644 --- a/src/main/java/lion/Externel/BackupSubServer.java +++ b/src/main/java/lion/Externel/BackupSubServer.java @@ -7,6 +7,7 @@ import lion.Http.HttpRequestHandler; import lion.Http.HttpRequestModel; import lion.Http.HttpResponseModel; import lion.Service.SubscriptionSnapshotStore; +import lion.Service.SubscriptionGateway; import lombok.extern.slf4j.Slf4j; import java.io.IOException; @@ -29,7 +30,7 @@ import java.util.Objects; *

{@link #run()} 保留为旧入口（{@code Main} 用它起线程），内部改为阻塞等待，直到线程被中断。 */ @Slf4j -public final class BackupSubServer implements Runnable { +public final class BackupSubServer implements Runnable, AutoCloseable { /** 公开 Key 的合法长度下限，过短直接按不存在处理，避免无谓查表。 */ private static final int MIN_KEY_LENGTH = 6; @@ -41,12 +42,20 @@ public final class BackupSubServer implements Runnable { private final SubscriptionSnapshotStore snapshotStore; private final int port; private final int workerCount; + private final SubscriptionGateway gateway; private volatile HttpFileServer server; public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) { + this(snapshotStore, port, workerCount, null); + } + + /** gateway 由同一宿主管理生命周期；null 保持历史快照模式。 */ + public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount, + SubscriptionGateway gateway) { this.snapshotStore = Objects.requireNonNull(snapshotStore); this.port = port; this.workerCount = Math.max(1, workerCount); + this.gateway = gateway; } /** 旧入口：绑定端口、开始接收请求，并阻塞到线程被中断。 */ @@ -66,17 +75,25 @@ public final class BackupSubServer implements Runnable { running.close(); } catch (IOException e) { log.error("备机订阅服务停止: {}", e.getMessage()); + } finally { + close(); } } /** 建好并启动服务；调用方负责 {@link HttpFileServer#close()}。 */ public HttpFileServer startServer() throws IOException { - HttpFileServer started = HttpFileServer.create(port, "/", handler(snapshotStore), workerCount); + HttpFileServer started = HttpFileServer.create(port, "/", gateway == null ? handler(snapshotStore) : gateway, workerCount); started.start(); server = started; return started; } + @Override + public void close() { + if (server != null) server.close(); + if (gateway != null) gateway.close(); + } + /** 当前运行中的外壳，便于宿主统一关闭；未启动时为 null。 */ public HttpFileServer server() { return server; @@ -103,7 +120,7 @@ public final class BackupSubServer implements Runnable { } } - String[] segments = path.split("/"); + String[] segments = path.split("/", -1); // "" / "sub" / / if (segments.length != 4 || !"sub".equals(segments[1]) || !(V2_CLIENT.equals(segments[2]) || CLASH_CLIENT.equals(segments[2]))) diff --git a/src/main/java/lion/Http/HttpFileServer.java b/src/main/java/lion/Http/HttpFileServer.java index 6e23535..9a01715 100644 --- a/src/main/java/lion/Http/HttpFileServer.java +++ b/src/main/java/lion/Http/HttpFileServer.java @@ -142,13 +142,17 @@ public final class HttpFileServer implements AutoCloseable { static HttpRequestModel toRequestModel(HttpExchange exchange) { String rawPath = exchange.getRequestURI().getRawPath(); String decodedPath = decodePath(rawPath); + Map requestHeaders = new LinkedHashMap<>(); + exchange.getRequestHeaders().forEach((key, values) -> { + if (!values.isEmpty()) requestHeaders.put(key, values.getFirst()); + }); return new HttpRequestModel( exchange.getRequestMethod(), rawPath, decodedPath, parseQuery(exchange.getRequestURI().getRawQuery()), exchange.getRequestHeaders().getFirst("Range"), - remoteAddress(exchange)); + remoteAddress(exchange), requestHeaders); } /** 对端 IP；取不到时返回 null，交由上层按既有策略处理。 */ @@ -218,7 +222,7 @@ public final class HttpFileServer implements AutoCloseable { if (head || bodyLength == 0) { // HEAD 与空正文：手动给出长度，再以 -1 结尾确保不写正文。 - if (head) + if (head && !headers.containsKey("Content-Length")) headers.set("Content-Length", Long.toString(bodyLength)); exchange.sendResponseHeaders(response.status(), -1); return; diff --git a/src/main/java/lion/Http/HttpRequestModel.java b/src/main/java/lion/Http/HttpRequestModel.java index 6556471..3d4dc5c 100644 --- a/src/main/java/lion/Http/HttpRequestModel.java +++ b/src/main/java/lion/Http/HttpRequestModel.java @@ -24,6 +24,7 @@ public final class HttpRequestModel { private final Map queryParameters; private final String rangeHeader; private final String clientAddress; + private final Map headers; public HttpRequestModel(String method, String rawPath, String decodedPath, Map queryParameters, String rangeHeader) { @@ -32,6 +33,12 @@ public final class HttpRequestModel { public HttpRequestModel(String method, String rawPath, String decodedPath, Map queryParameters, String rangeHeader, String clientAddress) { + this(method, rawPath, decodedPath, queryParameters, rangeHeader, clientAddress, Map.of()); + } + + public HttpRequestModel(String method, String rawPath, String decodedPath, + Map queryParameters, String rangeHeader, String clientAddress, + Map headers) { this.method = Objects.requireNonNull(method, "method"); this.rawPath = rawPath; this.decodedPath = decodedPath; @@ -40,6 +47,10 @@ public final class HttpRequestModel { : Collections.unmodifiableMap(new LinkedHashMap<>(queryParameters)); this.rangeHeader = rangeHeader; this.clientAddress = clientAddress; + Map normalized = new LinkedHashMap<>(); + if (headers != null) + headers.forEach((key, value) -> normalized.put(key.toLowerCase(java.util.Locale.ROOT), value)); + this.headers = Collections.unmodifiableMap(normalized); } public String method() { @@ -76,6 +87,11 @@ public final class HttpRequestModel { } /** 对端 IP；由外壳填充，直接驱动业务逻辑时可为 null。 */ + /** 单值请求头，名称大小写不敏感。 */ + public String header(String name) { + return headers.get(name.toLowerCase(java.util.Locale.ROOT)); + } + public String clientAddress() { return clientAddress; } diff --git a/src/main/java/lion/Main.java b/src/main/java/lion/Main.java index 06f392f..be42ee0 100644 --- a/src/main/java/lion/Main.java +++ b/src/main/java/lion/Main.java @@ -4,6 +4,8 @@ import io.netty.bootstrap.Bootstrap; import lion.Config.Config; import lion.Externel.BackupSubServer; import lion.Service.SubscriptionSnapshotStore; +import lion.Service.SubscriptionGateway; +import java.io.IOException; import lombok.extern.slf4j.Slf4j; import java.nio.file.Paths; @@ -18,8 +20,17 @@ public class Main { java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret, Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes); snapshotStore.load(); - new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers), - "subscription-backup-http").start(); + try { + SubscriptionGateway gateway = Config.subscriptionPrimaryUrl.isBlank() ? null + : new SubscriptionGateway(snapshotStore, Config.subscriptionPrimaryUrl, + Config.subscriptionSyncSecret, Paths.get(Config.subscriptionEventDir)); + BackupSubServer subscriptions = new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, + Config.subscriptionHttpWorkers, gateway); + Runtime.getRuntime().addShutdownHook(new Thread(subscriptions::close, "subscription-shutdown")); + new Thread(subscriptions, "subscription-backup-http").start(); + } catch (IOException failure) { + throw new IllegalStateException("无法初始化订阅事件持久化目录", failure); + } new Thread(() -> MultiThreadedHTTPServer.main(null)).start(); // 启用可靠任务协议：任务与待 ACK 报告持久化到独立目录，重启后恢复。 new storageNode(snapshotStore, Paths.get(Config.taskJournalDir), true); diff --git a/src/main/java/lion/Service/SubscriptionEventJournal.java b/src/main/java/lion/Service/SubscriptionEventJournal.java new file mode 100644 index 0000000..fce9360 --- /dev/null +++ b/src/main/java/lion/Service/SubscriptionEventJournal.java @@ -0,0 +1,135 @@ +package lion.Service; + +import com.fasterxml.jackson.databind.JsonNode; +import lion.CustomUtil; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.nio.file.attribute.PosixFilePermissions; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.TreeMap; +import java.util.UUID; + +/** 每个成功的备用 GET 先同步落盘；只在主站明确确认后删除。无公开 Key 和订阅正文。 */ +public final class SubscriptionEventJournal { + private static final long MAX_BYTES = 256L * 1024 * 1024; + private static final int MAX_EVENTS = 100_000; + private static final int MAX_EVENT_BYTES = 16 * 1024; + private static final long RESERVE_BYTES = 16L * 1024 * 1024; + private final Path directory; + private final TreeMap events = new TreeMap<>(); + private long bytes; + + public record Pending(String filename, String eventId, long occurredAt, byte[] json) { } + + public SubscriptionEventJournal(Path directory) throws IOException { + this.directory = directory; + Files.createDirectories(directory); + if (Files.isSymbolicLink(directory)) throw new IOException("Subscription journal cannot be a symlink"); + Files.setPosixFilePermissions(directory, PosixFilePermissions.fromString("rwx------")); + try (var files = Files.list(directory)) { + for (Path file : files.filter(p -> p.getFileName().toString().endsWith(".json")).toList()) { + if (!Files.isRegularFile(file, java.nio.file.LinkOption.NOFOLLOW_LINKS) + || Files.size(file) > MAX_EVENT_BYTES) + throw new IOException("Invalid subscription event file"); + byte[] json = Files.readAllBytes(file); + JsonNode value = CustomUtil.objectMapper.readTree(json); + Pending pending = pending(file.getFileName().toString(), value, json); + Files.setPosixFilePermissions(file, PosixFilePermissions.fromString("rw-------")); + events.put(pending.filename(), pending); + bytes += json.length; + } + } + } + + private static Pending pending(String filename, JsonNode value, byte[] json) throws IOException { + try { + String eventId = UUID.fromString(value.path("eventId").asText()).toString(); + long occurredAt = value.path("occurredAt").asLong(-1); + if (!value.isObject() || !value.path("occurredAt").isIntegralNumber() || occurredAt <= 0 + || !filename.equals(filename(occurredAt, eventId)) + || !value.path("keyHash").isTextual() || !value.path("keyHash").asText().matches("[a-f0-9]{64}") + || !value.path("client").isTextual() || !Set.of("v2", "cat").contains(value.path("client").asText()) + || !value.path("clientIp").isTextual() || value.path("clientIp").asText().length() > 45 + || !value.path("clientIp").asText().matches("[0-9a-fA-F:.]+") + || !value.path("userAgent").isTextual() || value.path("userAgent").asText().length() > 2048) + throw new IllegalArgumentException("Invalid event metadata"); + Set fields = Set.of("eventId", "keyHash", "client", "clientIp", "userAgent", "occurredAt"); + var fieldNames = value.fieldNames(); + while (fieldNames.hasNext()) if (!fields.contains(fieldNames.next())) + throw new IllegalArgumentException("Unknown event field"); + return new Pending(filename, eventId, occurredAt, json); + } catch (IllegalArgumentException failure) { + throw new IOException("Invalid subscription event metadata", failure); + } + } + + private static String filename(long time, String id) { + return String.format(java.util.Locale.ROOT, "%019d-%s.json", time, id); + } + + public synchronized void append(JsonNode event) throws IOException { + byte[] json = CustomUtil.objectMapper.writeValueAsBytes(event); + Pending pending = pending(filename(event.path("occurredAt").asLong(), event.path("eventId").asText()), event, json); + if (events.containsKey(pending.filename())) return; + if (json.length > MAX_EVENT_BYTES || events.size() >= MAX_EVENTS || bytes + json.length > MAX_BYTES + || Files.getFileStore(directory).getUsableSpace() < RESERVE_BYTES + json.length) + throw new IOException("Subscription event journal capacity exceeded"); + Path target = directory.resolve(pending.filename()); + Path temporary = directory.resolve(pending.filename() + ".tmp"); + try { + try (FileChannel channel = FileChannel.open(temporary, Set.of(StandardOpenOption.CREATE_NEW, + StandardOpenOption.WRITE), PosixFilePermissions.asFileAttribute( + PosixFilePermissions.fromString("rw-------")))) { + ByteBuffer buffer = ByteBuffer.wrap(json); + while (buffer.hasRemaining()) channel.write(buffer); + channel.force(true); + } + Files.move(temporary, target, StandardCopyOption.ATOMIC_MOVE); + events.put(pending.filename(), pending); + bytes += json.length; + forceDirectory(); + } finally { + Files.deleteIfExists(temporary); + } + } + + public synchronized List oldest(int limit) { + return new ArrayList<>(events.values().stream().limit(limit).toList()); + } + + /** 未确认或本批以外的 ID 绝不清理；目录同步失败时保留内存项，下轮继续幂等重试。 */ + public synchronized void acknowledge(List batch, Set acknowledged) throws IOException { + List deleted = new ArrayList<>(); + for (Pending pending : batch) { + if (acknowledged.contains(pending.eventId()) && events.containsKey(pending.filename())) { + Files.deleteIfExists(directory.resolve(pending.filename())); + deleted.add(pending); + } + } + if (deleted.isEmpty()) return; + forceDirectory(); + for (Pending pending : deleted) { + events.remove(pending.filename()); + bytes -= pending.json().length; + } + } + + private void forceDirectory() throws IOException { + try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) { + channel.force(true); + } + } + + public synchronized int size() { return events.size(); } + public synchronized long oldestAgeMillis() { + return events.isEmpty() ? 0 : Math.max(0, System.currentTimeMillis() - events.firstEntry().getValue().occurredAt()); + } +} diff --git a/src/main/java/lion/Service/SubscriptionGateway.java b/src/main/java/lion/Service/SubscriptionGateway.java new file mode 100644 index 0000000..584cec5 --- /dev/null +++ b/src/main/java/lion/Service/SubscriptionGateway.java @@ -0,0 +1,338 @@ +package lion.Service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import lion.CustomUtil; +import lion.Externel.BackupSubServer; +import lion.Http.HttpRequestHandler; +import lion.Http.HttpRequestModel; +import lion.Http.HttpResponseModel; +import lombok.extern.slf4j.Slf4j; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.io.IOException; +import java.net.InetAddress; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.time.Duration; +import java.util.Base64; +import java.util.HashSet; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Flow; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +/** storageNode 内置的主站优先订阅入口及离线访问记录回传器。 */ +@Slf4j +public final class SubscriptionGateway implements HttpRequestHandler, AutoCloseable { + private static final int MAX_RESPONSE_BYTES = 16 * 1024 * 1024; + private static final int MAX_BATCH_BYTES = 256 * 1024; + private final HttpRequestHandler local; + private final URI origin; + private final byte[] secret; + private final SubscriptionEventJournal journal; + private final HttpClient client; + private final Duration timeout; + private final long cooldownMillis; + private final ScheduledExecutorService uploader; + private final Object uploadLock = new Object(); + private long openUntil; + private long generation; + private boolean halfOpen; + private volatile long lastUploadAt; + private volatile String lastUploadState = "idle"; + private volatile boolean closed; + + public SubscriptionGateway(SubscriptionSnapshotStore store, String primaryUrl, String secret, Path eventDir) + throws IOException { + this(store, primaryUrl, secret, eventDir, Duration.ofSeconds(3), 10_000, true); + } + + /** 可配置时间仅供真实 HTTP 测试；生产固定 1s 连接/3s 总时限/10s 熔断。 */ + public SubscriptionGateway(SubscriptionSnapshotStore store, String primaryUrl, String secret, Path eventDir, + Duration timeout, long cooldownMillis, boolean backgroundUpload) throws IOException { + this.local = BackupSubServer.handler(store); + this.origin = validateOrigin(primaryUrl); + if (secret == null || secret.isBlank()) throw new IllegalArgumentException("Subscription gateway secret required"); + this.secret = secret.getBytes(StandardCharsets.UTF_8); + this.journal = new SubscriptionEventJournal(eventDir); + this.timeout = timeout; + this.cooldownMillis = cooldownMillis; + this.client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(1)) + .followRedirects(HttpClient.Redirect.NEVER).build(); + this.uploader = Executors.newSingleThreadScheduledExecutor(r -> { + Thread thread = new Thread(r, "subscription-events-upload"); + thread.setDaemon(true); + return thread; + }); + if (backgroundUpload) uploader.scheduleWithFixedDelay(this::uploadSafely, 10, 10, TimeUnit.SECONDS); + } + + private static URI validateOrigin(String value) { + URI uri = URI.create(value); + if (!("http".equals(uri.getScheme()) || "https".equals(uri.getScheme())) + || uri.getHost() == null || uri.getUserInfo() != null || uri.getQuery() != null + || uri.getFragment() != null || !(uri.getPath().isEmpty() || "/".equals(uri.getPath()))) + throw new IllegalArgumentException("SubscriptionPrimaryUrl must be a fixed HTTP(S) origin"); + return uri; + } + + @Override + public HttpResponseModel handle(HttpRequestModel request) { + String path = request.decodedPath(); + if ("/health/subscription-gateway".equals(path)) { + if (!HttpRequestModel.GET.equals(request.method()) && !request.isHead()) return HttpResponseModel.status(405); + try { + return HttpResponseModel.bytes(200, CustomUtil.objectMapper.writeValueAsBytes(status()), + "application/json; charset=utf-8", null, Map.of()); + } catch (IOException failure) { + return HttpResponseModel.status(500); + } + } + if (!HttpRequestModel.GET.equals(request.method()) && !request.isHead()) return local.handle(request); + String[] parts = path == null ? new String[0] : path.split("/", -1); + if (parts.length != 4 || !parts[0].isEmpty() || !"sub".equals(parts[1]) + || !("v2".equals(parts[2]) || "cat".equals(parts[2])) + || parts[3].length() < 6 || parts[3].length() > 512) + return local.handle(request); + // 内部路径不能成为 URI 目录跳转或带控制字符的请求。 + if (parts[3].codePoints().anyMatch(c -> Character.isISOControl(c))) return HttpResponseModel.status(404); + ObjectNode event; + try { + event = event(request, parts[2], parts[3]); + } catch (IOException failure) { + return HttpResponseModel.status(503); + } + Attempt attempt = acquire(); + if (attempt != null) { + try { + HttpResponseModel response = primary(request, parts, event); + if (response.status() < 500 && !(response.status() >= 300 && response.status() < 400)) { + recovered(attempt); + return response; + } + failed(attempt); + } catch (Exception failure) { + if (failure instanceof InterruptedException) Thread.currentThread().interrupt(); + failed(attempt); + // 不记录 URL/公开 key/密钥，失败原因只暴露类别。 + log.debug("订阅主站请求失败: {}", failure.getClass().getSimpleName()); + } + } + HttpResponseModel fallback = local.handle(request); + if (!request.isHead() && (fallback.status() == 200 || fallback.status() == 206)) { + try { + journal.append(event); + } catch (IOException failure) { + log.error("订阅离线记录无法持久化，拒绝本次更新: {}", failure.getClass().getSimpleName()); + return HttpResponseModel.text(503, "text/plain", "Subscription access journal unavailable"); + } + } + return fallback; + } + + private ObjectNode event(HttpRequestModel request, String format, String key) throws IOException { + ObjectNode event = CustomUtil.objectMapper.createObjectNode(); + event.put("eventId", UUID.randomUUID().toString()); + event.put("keyHash", sha256(key.getBytes(StandardCharsets.UTF_8))); + event.put("client", format); + event.put("clientIp", clientIp(request)); + String rawAgent = request.header("User-Agent"); + if (rawAgent == null) rawAgent = ""; + // HTTP/1 header值使用Latin-1。正常原始UA保持不变；移除控制字符， + // 非Latin-1字符替换为?，避免非法header导致每次请求都只能fallback。 + StringBuilder normalized = new StringBuilder(); + rawAgent.codePoints().filter(c -> c >= 32 && c != 127) + .limit(2048).forEach(c -> normalized.appendCodePoint(c <= 255 ? c : '?')); + String agent = normalized.toString(); + while (agent.getBytes(StandardCharsets.UTF_8).length > 2048) + agent = agent.substring(0, agent.length() - 1); + event.put("userAgent", agent); + event.put("occurredAt", System.currentTimeMillis()); + // 签名事件头的 base64url 必须 <=4096。保留能装下的原始前缀，避免切断代理对。 + while (Base64.getUrlEncoder().withoutPadding().encodeToString( + CustomUtil.objectMapper.writeValueAsBytes(event)).length() > 4096) { + agent = agent.substring(0, agent.offsetByCodePoints(agent.length(), -1)); + event.put("userAgent", agent); + } + return event; + } + + public static String clientIp(HttpRequestModel request) { + InetAddress remote = literalAddress(request.clientAddress()); + if (remote != null && remote.isLoopbackAddress()) { + InetAddress forwarded = literalAddress(request.header("X-Real-IP")); + if (forwarded != null) return forwarded.getHostAddress(); + } + return remote == null ? "0.0.0.0" : remote.getHostAddress(); + } + + private static InetAddress literalAddress(String value) { + if (value == null || value.length() > 45 || !value.matches("[0-9a-fA-F:.]+")) return null; + if (!value.contains(":")) { + if (!value.matches("[0-9]{1,3}(\\.[0-9]{1,3}){3}")) return null; + for (String part : value.split("\\.")) if (Integer.parseInt(part) > 255) return null; + } + try { return InetAddress.getByName(value); } catch (Exception ignored) { return null; } + } + + private HttpResponseModel primary(HttpRequestModel request, String[] parts, ObjectNode event) throws Exception { + URI target = endpoint("/sub/" + parts[2] + "/" + parts[3]); + HttpRequest.Builder builder = HttpRequest.newBuilder(target).timeout(timeout) + .method(request.isHead() ? "HEAD" : "GET", HttpRequest.BodyPublishers.noBody()) + .header("User-Agent", event.path("userAgent").asText()); + if (request.rangeHeader() != null) builder.header("Range", request.rangeHeader()); + // HEAD 也必须签名，使主站绕过普通访问记录分支；主站与节点均不记 HEAD。 + byte[] json = CustomUtil.objectMapper.writeValueAsBytes(event); + builder.header("X-Subscription-Event", Base64.getUrlEncoder().withoutPadding().encodeToString(json)); + sign(builder, json); + HttpResponse response = send(builder.build(), MAX_RESPONSE_BYTES); + Map headers = new LinkedHashMap<>(); + for (String name : List.of("Content-Range", "Accept-Ranges")) + response.headers().firstValue(name).ifPresent(value -> headers.put(name, value)); + if (request.isHead()) response.headers().firstValue("Content-Length") + .ifPresent(value -> headers.put("Content-Length", value)); + return HttpResponseModel.bytes(response.statusCode(), response.body(), + response.headers().firstValue("Content-Type").orElse("application/octet-stream"), null, headers); + } + + /** URI 构造器百分号编码单一路径；从不拼接用户 origin/query 或跟随 redirect。 */ + private URI endpoint(String path) throws URISyntaxException { + return new URI(origin.getScheme(), null, origin.getHost(), origin.getPort(), path, null, null); + } + + private void sign(HttpRequest.Builder builder, byte[] payload) throws Exception { + String timestamp = Long.toString(System.currentTimeMillis() / 1000); + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(secret, "HmacSHA256")); + String input = "v1\n" + timestamp + "\n" + sha256(payload); + builder.header("X-Subscription-Timestamp", timestamp) + .header("X-Subscription-Signature", HexFormat.of().formatHex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8)))); + } + + private static String sha256(byte[] bytes) { + try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)); } + catch (java.security.NoSuchAlgorithmException impossible) { throw new IllegalStateException(impossible); } + } + + /** sendAsync 的整次 future 包含正文消费，避免 ofInputStream 绕过读取时限。 */ + private HttpResponse send(HttpRequest request, int maxBytes) throws Exception { + CompletableFuture> future = client.sendAsync(request, + info -> new BoundedSubscriber(maxBytes)); + try { return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS); } + finally { if (!future.isDone()) future.cancel(true); } + } + + private static final class BoundedSubscriber implements HttpResponse.BodySubscriber { + private final HttpResponse.BodySubscriber delegate = HttpResponse.BodySubscribers.ofByteArray(); + private final int limit; + private int received; + private Flow.Subscription subscription; + BoundedSubscriber(int limit) { this.limit = limit; } + public CompletionStage getBody() { return delegate.getBody(); } + public void onSubscribe(Flow.Subscription value) { subscription = value; delegate.onSubscribe(value); } + public void onNext(List buffers) { + long size = buffers.stream().mapToLong(ByteBuffer::remaining).sum(); + if (size > limit - received) { + subscription.cancel(); + delegate.onError(new IOException("Subscription HTTP response exceeds limit")); + } else { received += (int) size; delegate.onNext(buffers); } + } + public void onError(Throwable error) { delegate.onError(error); } + public void onComplete() { delegate.onComplete(); } + } + + private record Attempt(long generation) { } + private synchronized Attempt acquire() { + if (openUntil == 0) return new Attempt(generation); + if (System.currentTimeMillis() < openUntil || halfOpen) return null; + halfOpen = true; + return new Attempt(generation); + } + private synchronized void failed(Attempt attempt) { + if (attempt.generation() != generation) return; + openUntil = System.currentTimeMillis() + cooldownMillis; + halfOpen = false; + generation++; + } + private synchronized void recovered(Attempt attempt) { + if (attempt.generation() != generation) return; + openUntil = 0; + halfOpen = false; + } + + public Map status() { + String circuit; + synchronized (this) { circuit = openUntil == 0 ? "closed" : halfOpen ? "half-open" : "open"; } + return Map.of("pendingCount", journal.size(), "oldestAgeMillis", journal.oldestAgeMillis(), + "circuit", circuit, "lastUploadAt", lastUploadAt, "lastUploadState", lastUploadState); + } + + private void uploadSafely() { + try { uploadOnce(); } + catch (Exception failure) { + lastUploadState = "failed"; + log.debug("订阅访问记录回传失败: {}", failure.getClass().getSimpleName()); + } + } + + /** 每轮最多一批、100条、256KiB；所有未明确确认的原始 eventId 均留盘。 */ + public void uploadOnce() throws Exception { + synchronized (uploadLock) { + if (closed) return; + List candidates = journal.oldest(100); + if (candidates.isEmpty()) { lastUploadState = "idle"; return; } + ObjectNode body = CustomUtil.objectMapper.createObjectNode(); + var array = body.putArray("events"); + java.util.ArrayList batch = new java.util.ArrayList<>(); + int total = 13; // {"events":[]} + 逗号安全余量 + for (var pending : candidates) { + if (total + pending.json().length + 1 > MAX_BATCH_BYTES) break; + array.add(CustomUtil.objectMapper.readTree(pending.json())); + batch.add(pending); + total += pending.json().length + 1; + } + if (batch.isEmpty()) throw new IOException("Subscription event exceeds upload batch limit"); + byte[] json = CustomUtil.objectMapper.writeValueAsBytes(body); + HttpRequest.Builder builder = HttpRequest.newBuilder(endpoint("/internal/subscription-events")) + .timeout(timeout).header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofByteArray(json)); + sign(builder, json); + HttpResponse response = send(builder.build(), 64 * 1024); + if (response.statusCode() != 200) throw new IOException("Subscription event upload rejected"); + JsonNode result = CustomUtil.objectMapper.readTree(response.body()); + if (result == null || !result.path("acknowledged").isArray()) throw new IOException("Invalid subscription event acknowledgement"); + Set ids = new HashSet<>(); + for (JsonNode id : result.path("acknowledged")) { + if (!id.isTextual()) throw new IOException("Invalid subscription event acknowledgement"); + ids.add(id.asText()); + } + journal.acknowledge(batch, ids); + lastUploadAt = System.currentTimeMillis(); + lastUploadState = "ok"; + } + } + + @Override + public void close() { + closed = true; + uploader.shutdownNow(); + client.close(); + } +} diff --git a/src/main/resources/config.properties b/src/main/resources/config.properties index 011f441..0837c6e 100644 --- a/src/main/resources/config.properties +++ b/src/main/resources/config.properties @@ -9,3 +9,9 @@ SubscriptionHttpWorkers=4 SubscriptionSocketTimeoutMs=10000 # 可靠任务协议的任务日志目录（独立于订阅数据目录，默认不接触订阅/测试生产目录） TaskJournalDir=/root/gallery/storageNode/task-journal + +# 主站优先网关（与订阅服务共用8889；空值关闭，保留快照直读） +# 生产设置 https://lionwebsite.xyz，仅允许固定origin，不含路径/认证信息。 +SubscriptionPrimaryUrl= +# 离线成功GET的耐久队列，独立于snapshot/task-journal，权限700/600。 +SubscriptionEventDir=/root/gallery/storageNode/subscription-events diff --git a/src/test/java/lion/ConfigTest.java b/src/test/java/lion/ConfigTest.java index 0c3386a..baf35f8 100644 --- a/src/test/java/lion/ConfigTest.java +++ b/src/test/java/lion/ConfigTest.java @@ -31,6 +31,8 @@ class ConfigTest { assertEquals(8889, Config.subscriptionHttpPort); assertEquals(4, Config.subscriptionHttpWorkers); assertEquals(10000, Config.subscriptionSocketTimeoutMs); + assertEquals("", Config.subscriptionPrimaryUrl); + assertEquals("/root/gallery/storageNode/subscription-events", Config.subscriptionEventDir); } @Test @@ -94,6 +96,18 @@ class ConfigTest { assertEquals("/root/gallery", Config.adminDownloadRoot); } + @Test + void gatewayConfigurationRequiresSecretAndHasIndependentEventDirectory(@TempDir Path root) throws Exception { + Path file = Files.writeString(root.resolve("gateway.properties"), + "SubscriptionPrimaryUrl=https://lionwebsite.xyz\nSubscriptionEventDir=/tmp/events\n"); + assertThrows(IllegalStateException.class, () -> Config.loadConfig(file.toString())); + Files.writeString(file, "SubscriptionPrimaryUrl=https://lionwebsite.xyz\n" + + "SubscriptionSyncSecret=gateway-secret\nSubscriptionEventDir=/tmp/events\n"); + Config.loadConfig(file.toString()); + assertEquals("https://lionwebsite.xyz", Config.subscriptionPrimaryUrl); + assertEquals("/tmp/events", Config.subscriptionEventDir); + } + /** * 启用订阅同步却没有密钥时必须拒绝启动。 * 否则备机会以空密钥运行，签名校验形同虚设。 diff --git a/src/test/java/lion/Service/SubscriptionGatewayTest.java b/src/test/java/lion/Service/SubscriptionGatewayTest.java new file mode 100644 index 0000000..7743edc --- /dev/null +++ b/src/test/java/lion/Service/SubscriptionGatewayTest.java @@ -0,0 +1,469 @@ +package lion.Service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import lion.CustomUtil; +import lion.Externel.BackupSubServer; +import lion.Http.HttpRequestModel; +import lion.Http.HttpResponseModel; +import lion.Message.Main.SubscriptionAccountSnapshot; +import lion.Message.Main.SubscriptionBindingSnapshot; +import lion.Message.Main.SubscriptionSnapshotMessage; +import lion.Message.Main.SubscriptionSnapshotPayload; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.io.ByteArrayOutputStream; +import java.net.InetSocketAddress; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Base64; +import java.util.HexFormat; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.zip.GZIPOutputStream; + +import static org.junit.jupiter.api.Assertions.*; + +class SubscriptionGatewayTest { + private static final String SECRET = "gateway-test-secret"; + private static final String KEY = "public-key-abcdef"; + private static final String LOCAL = "last synced snapshot"; + private static final HttpClient HTTP = HttpClient.newHttpClient(); + @TempDir Path root; + + @Test + void primaryPreferredAndSignedWithoutForwardingExternalAuth() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + HttpResponse response = h.request("GET", "User-Agent", "Clash/1.2", "X-Real-IP", "198.51.100.9", + "Authorization", "Bearer never-forward", "X-Subscription-Event", "attacker"); + assertEquals(200, response.statusCode()); + assertEquals("fresh primary", response.body()); + assertEquals(0, h.pending()); + JsonNode event = h.lastEvent.get(); + assertEquals(hash(KEY.getBytes(StandardCharsets.UTF_8)), event.path("keyHash").asText()); + assertEquals("Clash/1.2", event.path("userAgent").asText()); + assertEquals("198.51.100.9", event.path("clientIp").asText()); + assertEquals("v2", event.path("client").asText()); + assertNull(h.authorization.get()); + assertTrue(h.signatureValid); + assertFalse(new String(h.eventBytes.get(), StandardCharsets.UTF_8).contains(KEY)); + } + } + + @Test + void fallbackDurablyRecordsAndUploadAcknowledgementClears() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + h.primaryStatus.set(503); + assertEquals(LOCAL, h.request("GET", "User-Agent", "real-agent").body()); + assertEquals(1, h.pending()); + SubscriptionEventJournal recovered = new SubscriptionEventJournal(h.eventDir); + assertEquals(1, recovered.size(), "事件必须可以从磁盘恢复"); + String eventId = recovered.oldest(1).getFirst().eventId(); + h.gateway.uploadOnce(); + assertEquals(0, h.pending()); + assertEquals(eventId, h.uploaded.get().path("events").get(0).path("eventId").asText()); + assertEquals(0, new SubscriptionEventJournal(h.eventDir).size()); + assertTrue(h.signatureValid); + } + } + + @Test + void badUnknownAndMissingAcknowledgementsRetainStableIdsAcrossRestart() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + h.primaryStatus.set(503); + h.request("GET"); + String id = new SubscriptionEventJournal(h.eventDir).oldest(1).getFirst().eventId(); + h.ackMode.set(1); // 未知ID + h.gateway.uploadOnce(); + assertEquals(1, h.pending()); + h.ackMode.set(2); // 非法JSON + assertThrows(Exception.class, h.gateway::uploadOnce); + assertEquals(1, h.pending()); + h.ackMode.set(3); // 没有ack字段 + assertThrows(Exception.class, h.gateway::uploadOnce); + assertEquals(1, h.pending()); + h.gateway.close(); + try (SubscriptionGateway restarted = new SubscriptionGateway(h.store, h.origin, SECRET, h.eventDir, + Duration.ofMillis(500), 100, false)) { + assertEquals(1, ((Number) restarted.status().get("pendingCount")).intValue()); + h.ackMode.set(0); + restarted.uploadOnce(); + assertEquals(id, h.uploaded.get().path("events").get(0).path("eventId").asText()); + assertEquals(0, ((Number) restarted.status().get("pendingCount")).intValue()); + } + } + } + + @Test + void headAndNotFoundDoNotProduceSuccessfulRecordsOrFallback() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + h.primaryStatus.set(404); + HttpResponse missing = h.request("GET"); + assertEquals(404, missing.statusCode()); + assertEquals("not found", missing.body()); + assertEquals(0, h.pending()); + h.primaryStatus.set(503); + assertEquals(200, h.request("HEAD").statusCode()); + assertEquals(0, h.pending()); + assertEquals(1, h.primaryGets.get(), "HEAD 不得被记成GET"); + } + } + + @Test + void primaryHeadAndRangeHeadersSurviveGateway() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + HttpResponse head = h.request("HEAD"); + assertEquals("13", head.headers().firstValue("Content-Length").orElseThrow()); + assertEquals("", head.body()); + assertEquals(0, h.pending()); + assertNotNull(h.lastEvent.get(), "HEAD 必须附签名，主站才不会按普通请求记录"); + assertTrue(h.signatureValid); + HttpResponse range = h.request("GET", "Range", "bytes=0-4"); + assertEquals(206, range.statusCode()); + assertEquals("fresh", range.body()); + assertEquals("bytes 0-4/13", range.headers().firstValue("Content-Range").orElseThrow()); + assertEquals("bytes", range.headers().firstValue("Accept-Ranges").orElseThrow()); + assertEquals("bytes=0-4", h.lastRange.get()); + } + } + + @Test + void primaryRecordedBeforeTimeoutReplaysSameEventId() throws Exception { + try (Harness h = new Harness(root, 200, 100)) { + h.primaryDelay.set(650); + long start = System.nanoTime(); + HttpResponse response = h.request("GET"); + assertEquals(LOCAL, response.body()); + assertTrue(Duration.ofNanos(System.nanoTime() - start).toMillis() < 1500); + String id = h.lastEvent.get().path("eventId").asText(); + assertEquals(id, new SubscriptionEventJournal(h.eventDir).oldest(1).getFirst().eventId()); + h.gateway.uploadOnce(); + assertEquals(id, h.uploaded.get().path("events").get(0).path("eventId").asText()); + assertEquals(0, h.pending()); + } + } + + @Test + void circuitSkipsFailedOriginThenRecoversToPrimary() throws Exception { + try (Harness h = new Harness(root, 500, 150)) { + h.primaryStatus.set(503); + assertEquals(LOCAL, h.request("GET").body()); + assertEquals(LOCAL, h.request("GET").body()); + assertEquals(1, h.primaryGets.get()); + assertEquals("open", h.gateway.status().get("circuit")); + h.primaryStatus.set(200); + Thread.sleep(180); + assertEquals("fresh primary", h.request("GET").body()); + assertEquals("closed", h.gateway.status().get("circuit")); + assertEquals("fresh primary", h.request("GET").body()); + assertEquals(3, h.primaryGets.get()); + } + } + + @Test + void halfOpenAllowsOnlyOneProbeAndBackgroundUploadDoesNotBlockRequests() throws Exception { + try (Harness h = new Harness(root, 1000, 100)) { + h.primaryStatus.set(503); + h.request("GET"); + Thread.sleep(130); + h.primaryStatus.set(200); + h.probeGate = new CountDownLatch(1); + CompletableFuture> probe = CompletableFuture.supplyAsync(() -> { + try { return h.request("GET"); } catch (Exception e) { throw new RuntimeException(e); } + }); + assertTrue(h.probeEntered.await(1, TimeUnit.SECONDS)); + assertEquals(LOCAL, h.request("GET").body()); + assertEquals(2, h.primaryGets.get()); + h.probeGate.countDown(); + assertEquals("fresh primary", probe.get(2, TimeUnit.SECONDS).body()); + + h.uploadDelay.set(700); + CompletableFuture upload = CompletableFuture.runAsync(() -> { + try { h.gateway.uploadOnce(); } catch (Exception e) { throw new RuntimeException(e); } + }); + assertTrue(h.uploadEntered.await(1, TimeUnit.SECONDS)); + long start = System.nanoTime(); + assertEquals("fresh primary", h.request("GET").body()); + assertTrue(Duration.ofNanos(System.nanoTime() - start).toMillis() < 500, + "上传不得持有请求路径使用的熔断器锁"); + upload.get(2, TimeUnit.SECONDS); + } + } + + @Test + void diskFailureReturns503InsteadOfUnrecordedSuccess() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + Files.delete(h.eventDir); + Files.writeString(h.eventDir, "not a directory"); + h.primaryStatus.set(503); + assertEquals(503, h.request("GET").statusCode()); + assertEquals(0, h.pending()); + } + } + + @Test + void onlyLoopbackCanSupplyRealIpAndXffIsAlwaysIgnored() { + assertEquals("203.0.113.12", SubscriptionGateway.clientIp(model("GET", KEY, "203.0.113.12", + Map.of("X-Real-IP", "198.51.100.8", "X-Forwarded-For", "198.51.100.7")))); + assertEquals("198.51.100.8", SubscriptionGateway.clientIp(model("GET", KEY, "127.0.0.1", + Map.of("X-Real-IP", "198.51.100.8", "X-Forwarded-For", "198.51.100.7")))); + assertEquals("127.0.0.1", SubscriptionGateway.clientIp(model("GET", KEY, "127.0.0.1", + Map.of("X-Real-IP", "attacker.example", "X-Forwarded-For", "198.51.100.7")))); + } + + @Test + void encodedKeyNeverChangesOriginAndInvalidPathsNeverContactPrimary() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + String key = "safe+key?next=%23#fragment"; + assertEquals(200, h.gateway.handle(model("GET", key, "203.0.113.1", Map.of())).status()); + assertEquals("/sub/v2/" + key, h.lastPath.get()); + for (String path : List.of("/sub/v2/abc", "/sub/v2/abcdef/", "/sub/v2/abc/def", "/other/abcdef")) { + assertEquals(404, h.gateway.handle(new HttpRequestModel("GET", path, path, Map.of(), null)).status()); + } + assertEquals(1, h.primaryGets.get()); + assertThrows(IllegalArgumentException.class, () -> new SubscriptionGateway(h.store, + h.origin + "/arbitrary/path", SECRET, root.resolve("bad"))); + } + } + + @Test + void uploadBatchRemainsUnderBodyLimitWithEscapedAgents() throws Exception { + try (Harness h = new Harness(root, 500, 10000)) { + h.primaryStatus.set(503); + String agent = "\\".repeat(2048); + for (int i = 0; i < 100; i++) { + assertEquals(200, h.gateway.handle(model("GET", KEY, "203.0.113.9", Map.of("User-Agent", agent))).status()); + } + assertEquals(100, h.pending()); + h.gateway.uploadOnce(); + assertTrue(h.uploadBodyBytes.get() <= 256 * 1024); + assertTrue(h.pending() > 0, "100个大UA必须拆批"); + assertTrue(h.pending() < 100); + for (var pending : new SubscriptionEventJournal(h.eventDir).oldest(100)) { + assertTrue(Base64.getUrlEncoder().withoutPadding().encodeToString(pending.json()).length() <= 4096); + } + } + } + + @Test + void userAgentNormalizationKeepsHttpHeadersValidAndWithinBudget() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + assertEquals(200, h.gateway.handle(model("GET", KEY, "203.0.113.1", + Map.of("User-Agent", "猫😀\r\nAgent\t"))).status()); + assertEquals("??Agent", h.lastEvent.get().path("userAgent").asText()); + assertEquals(0, h.pending(), "规范后应正常访问主站而非仅fallback"); + assertEquals(200, h.gateway.handle(model("GET", KEY, "203.0.113.1", + Map.of("User-Agent", "é".repeat(2048)))).status()); + assertTrue(h.lastEvent.get().path("userAgent").asText().getBytes(StandardCharsets.UTF_8).length <= 2048); + assertTrue(h.eventHeaderLength.get() <= 4096); + } + } + + @Test + void primaryRedirectIsNotFollowedAndFallsBack() throws Exception { + try (Harness h = new Harness(root, 500, 100)) { + h.primaryStatus.set(302); + assertEquals(LOCAL, h.request("GET").body()); + assertEquals(1, h.primaryGets.get()); + assertEquals(1, h.pending()); + } + } + + @Test + void oversizedOrIncompletePrimaryBodiesFallBack() throws Exception { + try (Harness h = new Harness(root, 1000, 1)) { + h.primaryStatus.set(601); // 声明较长正文，只写一个字节 + assertEquals(LOCAL, h.request("GET").body()); + Thread.sleep(10); + h.primaryStatus.set(602); // 正文超过16MiB + assertEquals(LOCAL, h.request("GET").body()); + assertEquals(2, h.pending()); + } + } + + private static HttpRequestModel model(String method, String key, String peer, Map headers) { + String path = "/sub/v2/" + key; + return new HttpRequestModel(method, path, path, Map.of(), null, peer, headers); + } + + private static final class Harness implements AutoCloseable { + final HttpServer primary; + final java.util.concurrent.ExecutorService executor = Executors.newCachedThreadPool(); + final SubscriptionSnapshotStore store; + final SubscriptionGateway gateway; + final BackupSubServer backup; + final Path eventDir; + final String origin; + final AtomicInteger primaryStatus = new AtomicInteger(200); + final AtomicInteger primaryGets = new AtomicInteger(); + final AtomicInteger primaryDelay = new AtomicInteger(); + final AtomicInteger uploadDelay = new AtomicInteger(); + final AtomicInteger ackMode = new AtomicInteger(); + final AtomicInteger uploadBodyBytes = new AtomicInteger(); + final AtomicInteger eventHeaderLength = new AtomicInteger(); + final AtomicReference lastEvent = new AtomicReference<>(); + final AtomicReference uploaded = new AtomicReference<>(); + final AtomicReference eventBytes = new AtomicReference<>(); + final AtomicReference authorization = new AtomicReference<>(); + final AtomicReference lastRange = new AtomicReference<>(); + final AtomicReference lastPath = new AtomicReference<>(); + final CountDownLatch probeEntered = new CountDownLatch(1); + final CountDownLatch uploadEntered = new CountDownLatch(1); + volatile CountDownLatch probeGate; + volatile boolean signatureValid; + + Harness(Path root, int timeoutMs, long cooldownMs) throws Exception { + eventDir = root.resolve("events"); + store = readyStore(root.resolve("snapshot")); + primary = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 16); + primary.setExecutor(executor); + primary.createContext("/", this::serve); + primary.start(); + origin = "http://127.0.0.1:" + primary.getAddress().getPort(); + gateway = new SubscriptionGateway(store, origin, SECRET, eventDir, + Duration.ofMillis(timeoutMs), cooldownMs, false); + backup = new BackupSubServer(store, 0, 4, gateway); + backup.startServer(); + } + + void serve(HttpExchange exchange) { + try (exchange) { + if (exchange.getRequestURI().getPath().equals("/internal/subscription-events")) { + byte[] bytes = exchange.getRequestBody().readAllBytes(); + uploadBodyBytes.set(bytes.length); + signatureValid = signatureValid(exchange, bytes); + JsonNode value = CustomUtil.objectMapper.readTree(bytes); + uploaded.set(value); + uploadEntered.countDown(); + Thread.sleep(uploadDelay.get()); + String body; + if (ackMode.get() == 1) body = "{\"acknowledged\":[\"unknown-id\"]}"; + else if (ackMode.get() == 2) body = "invalid-json"; + else if (ackMode.get() == 3) body = "{}"; + else { + List ids = new ArrayList<>(); + value.path("events").forEach(event -> ids.add(event.path("eventId").asText())); + body = CustomUtil.objectMapper.writeValueAsString(Map.of("acknowledged", ids)); + } + respond(exchange, 200, body); + return; + } + lastPath.set(exchange.getRequestURI().getPath()); + boolean head = "HEAD".equals(exchange.getRequestMethod()); + if (!head) primaryGets.incrementAndGet(); + authorization.set(exchange.getRequestHeaders().getFirst("Authorization")); + lastRange.set(exchange.getRequestHeaders().getFirst("Range")); + String encoded = exchange.getRequestHeaders().getFirst("X-Subscription-Event"); + if (encoded != null) { + eventHeaderLength.set(encoded.length()); + byte[] bytes = Base64.getUrlDecoder().decode(encoded); + eventBytes.set(bytes); + lastEvent.set(CustomUtil.objectMapper.readTree(bytes)); + signatureValid = signatureValid(exchange, bytes); + } + if (probeGate != null) { probeEntered.countDown(); probeGate.await(2, TimeUnit.SECONDS); } + Thread.sleep(primaryDelay.get()); + int status = primaryStatus.get(); + if (status == 601) { + exchange.sendResponseHeaders(200, 100); + exchange.getResponseBody().write('a'); + return; + } + if (status == 602) { + exchange.sendResponseHeaders(200, 16 * 1024 * 1024 + 1); + byte[] chunk = new byte[64 * 1024]; + for (int i = 0; i < 256; i++) exchange.getResponseBody().write(chunk); + exchange.getResponseBody().write('z'); + return; + } + if (status == 302) exchange.getResponseHeaders().set("Location", origin + "/forbidden-redirect"); + exchange.getResponseHeaders().set("Content-Type", "text/plain"); + exchange.getResponseHeaders().set("Accept-Ranges", "bytes"); + if (head) { + exchange.getResponseHeaders().set("Content-Length", "13"); + exchange.sendResponseHeaders(status, -1); + } else if (status == 200 && lastRange.get() != null) { + exchange.getResponseHeaders().set("Content-Range", "bytes 0-4/13"); + respond(exchange, 206, "fresh"); + } else respond(exchange, status, status == 404 ? "not found" : "fresh primary"); + } catch (Exception ignored) { /* 断连/短正文/超时正是测试用例 */ } + } + + HttpResponse request(String method, String... headers) throws Exception { + var builder = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + backup.server().port() + "/sub/v2/" + KEY)) + .timeout(Duration.ofSeconds(5)).method(method, HttpRequest.BodyPublishers.noBody()); + if (headers.length > 0) builder.headers(headers); + return HTTP.send(builder.build(), HttpResponse.BodyHandlers.ofString()); + } + int pending() { return ((Number) gateway.status().get("pendingCount")).intValue(); } + @Override public void close() { backup.close(); primary.stop(0); executor.shutdownNow(); } + } + + private static void respond(HttpExchange exchange, int code, String text) throws Exception { + byte[] bytes = text.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(code, bytes.length); + exchange.getResponseBody().write(bytes); + } + private static boolean signatureValid(HttpExchange exchange, byte[] bytes) throws Exception { + String timestamp = exchange.getRequestHeaders().getFirst("X-Subscription-Timestamp"); + String expected = signature("v1\n" + timestamp + "\n" + hash(bytes)); + return expected.equals(exchange.getRequestHeaders().getFirst("X-Subscription-Signature")); + } + private static String signature(String input) throws Exception { + Mac mac = Mac.getInstance("HmacSHA256"); + mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256")); + return HexFormat.of().formatHex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8))); + } + private static String hash(byte[] bytes) throws Exception { + return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)); + } + private static SubscriptionSnapshotStore readyStore(Path path) throws Exception { + var account = new SubscriptionAccountSnapshot(); + account.setAccountId(1); + account.setEnabled(true); + byte[] content = LOCAL.getBytes(StandardCharsets.UTF_8); + account.setV2ContentBase64(Base64.getEncoder().encodeToString(content)); + account.setV2Sha256(hash(content)); + account.setClashContentBase64(Base64.getEncoder().encodeToString(content)); + account.setClashSha256(hash(content)); + var binding = new SubscriptionBindingSnapshot(); + binding.setAccountId(1); + binding.setPublicKeySha256(hash(KEY.getBytes(StandardCharsets.UTF_8))); + var payload = new SubscriptionSnapshotPayload(); + payload.setSchemaVersion(1); + payload.setAccounts(List.of(account)); + payload.setBindings(List.of(binding)); + byte[] json = CustomUtil.objectMapper.writeValueAsBytes(payload); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + try (GZIPOutputStream gzip = new GZIPOutputStream(out)) { gzip.write(json); } + byte[] zipped = out.toByteArray(); + var message = new SubscriptionSnapshotMessage(); + message.setSchemaVersion(1); + message.setRevision(hash(json)); + message.setGeneratedAt(System.currentTimeMillis()); + message.setPayloadBase64(Base64.getEncoder().encodeToString(zipped)); + message.setPayloadSha256(hash(zipped)); + message.setSignature(signature("1\n" + message.getRevision() + "\n" + message.getGeneratedAt() + "\n" + message.getPayloadSha256())); + var store = new SubscriptionSnapshotStore(path, SECRET, 3600, 1024 * 1024); + assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code()); + return store; + } +}