From 7305629f57c6f9f53b081936a9edf685a8d2eb74 Mon Sep 17 00:00:00 2001 From: wangzhizhou <824219521@qq.com> Date: Tue, 27 Oct 2026 07:40:44 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(stream):=20=E6=8E=A7=E5=88=B6=E5=8F=B0?= =?UTF-8?q?=E5=AE=9E=E6=97=B6=E6=B5=81=EF=BC=88Engine.IO=20v4=20+=20Socket?= =?UTF-8?q?.IO=20v5=20=E6=9C=80=E5=B0=8F=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 控制台原来只能 3 秒轮询 `outputlog`（纯文本尾部），实时性差、还浪费请求。面板本身有实时流： `POST /api/protected_instance/stream_channel` 换一个一次性口令，然后连 `addr` 上的 socket.io， 节点的 stdout 就会一条条推过来（面板前端就是这么做的）。 **`MCSMAPI`（生成客户端侧）** - spec 新增 `POST /api/protected_instance/stream_channel`（`getStreamChannel`）： 响应同时声明外壳 `{status,data,time}` 与裸形状（面板 10.x 实际是前者），`text/plain` 也收； - `MCSMStreamChannel`：把两种响应形状抹平成"连哪 + 口令 + 前缀 + 地址重映射"， 还支持从纯文本 JSON 解析（面板的老毛病：JSON 走 `text/plain`）。 **`MCSMStream`（新 product，无第三方依赖）** - `MCSMStreamEndpoint`：把 `addr` 解析成 socket.io URL。规则照抄面板前端 `parseForwardAddress`：显式协议优先，否则跟随面板（https→wss）； **`127.0.0.1`/`localhost` 换成本面板主机名**（面板返回的是节点视角地址，客户端连不上）； `prefix` 拼进出 `/socket.io` 路径。 - `SocketIOFrame`：Engine.IO v4 / Socket.IO v5 帧编解码（`0` open、`2/3` ping/pong、 `40/41` 连接、`42` 事件、`44` 错误）。事件参数保留**原始 JSON 文本**而不是 `[String: Any]` （`Any` 既非 `Equatable` 也非 `Sendable`），取值走 `authAccepted` / `stdoutText` / `instanceLifecycle`；编码用 `.withoutEscapingSlashes`，`stream/auth` 不会被写成 `stream\/auth`。 - `MCSMStreamClient`（actor）：`URLSessionWebSocketTask` 跑一次完整连接 （open → `40` → `stream/auth` → `stream/detail` → 转发 `instance/stdout`）， 自动回 pong（不回会被面板 25 秒后踢掉），握手阶段带超时。 **口令是一次性的**，所以重连要重新申请 → 构造时给的是 `makeEndpoint` 闭包而不是固定 URL； 重试用指数退避，全失败不崩，发 `.fallback(reason:)` 让调用方退回轮询 （`dataPlane=direct` 且节点在 NAT 后面时直连本来就不可达，这是正常路径）。 另有 `sendCommand(_:)`（走 `stream/input`），本包暂不使用，留给控制台"低延迟发命令"。 **测试 20 条**（`swift test` 本地全绿） - 帧编解码：控制包、事件取值、怪帧不崩（空串/半截 JSON/二进制前缀/无参数）、编码形状； - 地址解析：proxy/direct/回环替换/前缀/显式协议/http 面板/非法地址； - 通道模型：外壳、裸形状、纯文本 JSON、缺字段； - **端到端**：测试里起一个**真的 WebSocket 服务端**（Network.framework + `NWProtocolWebSocket`， 不依赖 node/npm），跑通「鉴权通过后收输出」「口令被拒 → 重试用尽 → `.fallback`」 「服务端 ping 要回 pong」「实例退出事件透传」「连不上不崩」。 --- Package.swift | 13 + Sources/MCSMAPI/StreamChannel.swift | 126 +++++++++ Sources/MCSMAPI/openapi.yml | 128 +++++++++ Sources/MCSMStream/MCSMStreamClient.swift | 262 ++++++++++++++++++ Sources/MCSMStream/SocketIOFrame.swift | 177 ++++++++++++ Sources/MCSMStream/StreamEndpoint.swift | 115 ++++++++ .../MCSMStreamTests/FakeSocketIOServer.swift | 150 ++++++++++ .../MCSMStreamClientTests.swift | 147 ++++++++++ .../MCSMStreamTests/SocketIOFrameTests.swift | 81 ++++++ .../MCSMStreamTests/StreamEndpointTests.swift | 128 +++++++++ 10 files changed, 1327 insertions(+) create mode 100644 Sources/MCSMAPI/StreamChannel.swift create mode 100644 Sources/MCSMStream/MCSMStreamClient.swift create mode 100644 Sources/MCSMStream/SocketIOFrame.swift create mode 100644 Sources/MCSMStream/StreamEndpoint.swift create mode 100644 Tests/MCSMStreamTests/FakeSocketIOServer.swift create mode 100644 Tests/MCSMStreamTests/MCSMStreamClientTests.swift create mode 100644 Tests/MCSMStreamTests/SocketIOFrameTests.swift create mode 100644 Tests/MCSMStreamTests/StreamEndpointTests.swift diff --git a/Package.swift b/Package.swift index ef1b34b..7615983 100644 --- a/Package.swift +++ b/Package.swift @@ -14,6 +14,7 @@ let package = Package( ], products: [ .library(name: "MCSMAPI", targets: ["MCSMAPI"]), + .library(name: "MCSMStream", targets: ["MCSMStream"]), ], dependencies: [ .package(url: "https://github.com/apple/swift-openapi-generator.git", from: "1.11.0"), @@ -34,6 +35,18 @@ let package = Package( .plugin(name: "OpenAPIGenerator", package: "swift-openapi-generator"), ] ), + .target( + name: "MCSMStream", + dependencies: ["MCSMAPI"] + ), + .testTarget( + name: "MCSMStreamTests", + dependencies: [ + "MCSMStream", + "MCSMAPI", + .product(name: "Testing", package: "swift-testing"), + ] + ), .testTarget( name: "MCSMAPITests", dependencies: [ diff --git a/Sources/MCSMAPI/StreamChannel.swift b/Sources/MCSMAPI/StreamChannel.swift new file mode 100644 index 0000000..46e4c31 --- /dev/null +++ b/Sources/MCSMAPI/StreamChannel.swift @@ -0,0 +1,126 @@ +// +// StreamChannel.swift +// MCSMAPI +// +// `POST /api/protected_instance/stream_channel` 的**业务视图**：把生成的响应模型 +// （带/不带 `{status,data,time}` 外壳两种形状）抹平成"连哪、用什么口令"。 +// +// 为什么需要这层：面板 direct / proxy 两种 `dataPlane` 返回的 `addr` 含义不同 +// （节点地址 / 面板自身地址），但客户端接下来要做的事**完全一样** —— +// 连 `addr` + `prefix`，发 `stream/auth`。所以这里只做归一化，不做模式判断。 +// + +import Foundation + +/// 一条控制台实时流通道。 +public struct MCSMStreamChannel: Equatable, Sendable { + + /// `host[:port]` 或 `ws(s)://host[:port][/path]`（原样，未解析）。 + public let address: String + /// 一次性口令：direct 模式由节点签发，proxy 模式是面板签发的网关令牌。 + public let password: String + /// socket.io 的路径前缀（proxy 模式为空；direct 模式跟随节点配置）。 + public let prefix: String + /// 面板的地址重映射（用不上就空）。 + public let remoteMappings: [MCSMRemoteMapping] + + public init(address: String, password: String, prefix: String = "", remoteMappings: [MCSMRemoteMapping] = []) { + self.address = address + self.password = password + self.prefix = prefix + self.remoteMappings = remoteMappings + } +} + +/// 面板的地址重映射：`from` 是浏览器看到的地址，`to` 是实际可连的地址。 +public struct MCSMRemoteMapping: Equatable, Sendable { + + public struct Endpoint: Equatable, Sendable { + public let address: String + public let prefix: String + + public init(address: String, prefix: String = "") { + self.address = address + self.prefix = prefix + } + } + + public let from: Endpoint + public let to: Endpoint + + public init(from: Endpoint, to: Endpoint) { + self.from = from + self.to = to + } +} + +extension MCSMStreamChannel { + + /// 从 `POST /api/protected_instance/stream_channel` 的响应里取通道。 + /// + /// 面板 10.x 实际返回带外壳的形状；也兼容不套外壳（顶层直接有 `addr`/`password`）。 + /// `addr` 或 `password` 缺失就返回 `nil`（说明面板版本太老，调用方回退轮询）。 + public init?(_ response: Components.Schemas.StreamChannelResponse) { + if let payload = response.data, + let channel = Self(validating: payload.addr, password: payload.password, + prefix: payload.prefix ?? "", + remoteMappings: Self.mappings(payload.remoteMappings)) { + self = channel + return + } + guard let address = response.addr, let password = response.password, + let channel = Self(validating: address, password: password, prefix: response.prefix ?? "", + remoteMappings: Self.mappings(response.remoteMappings)) + else { return nil } + self = channel + } + + /// 纯文本响应体（面板有时用 `text/plain` 送同一份 JSON）也能解。 + public init?(text: String) { + guard let data = text.data(using: .utf8), + let object = try? JSONSerialization.jsonObject(with: data) as? [String: Any] + else { return nil } + // 先剥外壳，再看顶层。 + let body = (object["data"] as? [String: Any]) ?? object + guard let address = body["addr"] as? String, let password = body["password"] as? String else { + return nil + } + guard let channel = Self(validating: address, password: password, + prefix: body["prefix"] as? String ?? "", + remoteMappings: MCSMRemoteMapping.parse(body["remoteMappings"])) + else { return nil } + self = channel + } + + private init?(validating address: String, password: String, prefix: String, remoteMappings: [MCSMRemoteMapping]?) { + guard !address.isEmpty, !password.isEmpty else { return nil } + self.init(address: address, password: password, prefix: prefix, remoteMappings: remoteMappings ?? []) + } + + private static func mappings(_ entries: [Components.Schemas.RemoteMappingEntry]?) -> [MCSMRemoteMapping] { + (entries ?? []).compactMap { entry in + guard let from = entry.from, let to = entry.to else { return nil } + return MCSMRemoteMapping( + from: .init(address: from.addr ?? "", prefix: from.prefix ?? ""), + to: .init(address: to.addr ?? "", prefix: to.prefix ?? "") + ) + } + } +} + +extension MCSMRemoteMapping { + + /// 从任意 JSON 值解析（`init?(text:)` 用；形状不对就当没有）。 + static func parse(_ value: Any?) -> [MCSMRemoteMapping] { + guard let array = value as? [[String: Any]] else { return [] } + return array.compactMap { entry in + guard let from = entry["from"] as? [String: Any], let to = entry["to"] as? [String: Any], + let fromAddr = from["addr"] as? String, let toAddr = to["addr"] as? String + else { return nil } + return MCSMRemoteMapping( + from: .init(address: fromAddr, prefix: from["prefix"] as? String ?? ""), + to: .init(address: toAddr, prefix: to["prefix"] as? String ?? "") + ) + } + } +} diff --git a/Sources/MCSMAPI/openapi.yml b/Sources/MCSMAPI/openapi.yml index 1f6dd5a..511ba45 100644 --- a/Sources/MCSMAPI/openapi.yml +++ b/Sources/MCSMAPI/openapi.yml @@ -461,6 +461,68 @@ paths: schema: type: string + /api/protected_instance/stream_channel: + post: + operationId: getStreamChannel + summary: 申请控制台实时流通道（节点地址 + 一次性口令） + description: | + 面板「实时终端」的第一步：拿到**往哪儿连**和**一次性口令**，然后连那个地址上的 + socket.io，发 `stream/auth` 换到真正的输出流。比轮询 `outputlog` 实时得多 + （面板内部就是把节点的 stdout 一条条推过来）。 + + 返回体的形状取决于面板的 `dataPlane` 配置（面板 10.18+）： + + - `direct`（默认）：`addr` 是**节点（守护进程）的地址**，客户端直连节点； + 节点在 NAT 后面时这个地址从公网不可达 —— 连不上就回退轮询。 + - `proxy`：`addr` 是**面板自己的对外地址**，`prefix` 为空，`password` 是面板签发的 + 网关令牌，由面板把 socket.io 中转给节点。公网部署通常用这个模式。 + + 两种模式下**客户端代码一样**：连 `addr` + `prefix`，发 `stream/auth`。 + + **普通用户也能调用**（同 `/protected_instance` 一组，只要实例属于你）。 + 口令是一次性的、有有效期；断开重连要重新申请（本包不缓存）。 + parameters: + - name: daemonId + in: query + required: true + schema: + type: string + - name: uuid + in: query + required: true + description: 实例 UUID（**不是**昵称） + schema: + type: string + responses: + "200": + description: 通道信息（带 `{status,data,time}` 外壳） + content: + application/json: + schema: + $ref: "#/components/schemas/StreamChannelResponse" + # 同其他接口：面板协议中间件会把体包成 JSON 但 Content-Type 有时还是 text/plain。 + text/plain: + schema: + type: string + "403": + description: API Key 权限不足 / 未开启 API Key 功能 / 不是这台实例的主人 + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + text/plain: + schema: + type: string + "500": + description: 面板内部错误 / 节点未连接 + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + text/plain: + schema: + type: string + components: securitySchemes: apiKeyHeader: @@ -472,6 +534,72 @@ components: 面板也接受 `?apikey=` 查询参数；本包用 header（不落 URL、不进日志）。 schemas: + StreamChannelResponse: + type: object + description: | + `POST /api/protected_instance/stream_channel` 的响应。 + + 两种形状都要能解：带外壳 `{status, data: {…}, time}`（面板 10.x 实际行为）， + 以及不套外壳时直接把 `addr` / `password` 放在顶层。 + properties: + status: + type: integer + time: + type: integer + format: int64 + data: + $ref: "#/components/schemas/StreamChannelPayload" + addr: + type: string + password: + type: string + prefix: + type: string + remoteMappings: + type: array + items: + $ref: "#/components/schemas/RemoteMappingEntry" + + StreamChannelPayload: + type: object + description: 通道本体：连哪、用什么口令。 + required: + - addr + - password + properties: + addr: + type: string + description: | + `host[:port]` 或 `ws(s)://host[:port][/path]`。 + `127.0.0.1` / `localhost` 开头的地址要换成本面板的主机名（节点和面板常在同一台机器）。 + password: + type: string + description: 一次性口令（direct 模式是节点签发的，proxy 模式是面板签发的网关令牌）。 + prefix: + type: string + description: socket.io 的路径前缀（proxy 模式为空；direct 模式跟随节点配置，如 `/daemon`）。 + remoteMappings: + type: array + items: + $ref: "#/components/schemas/RemoteMappingEntry" + + RemoteMappingEntry: + type: object + description: 面板的地址重映射（节点地址 → 浏览器可访问地址）。用不上时直接忽略。 + properties: + from: + $ref: "#/components/schemas/RemoteMappingEndpoint" + to: + $ref: "#/components/schemas/RemoteMappingEndpoint" + + RemoteMappingEndpoint: + type: object + properties: + addr: + type: string + prefix: + type: string + ErrorResponse: type: object required: diff --git a/Sources/MCSMStream/MCSMStreamClient.swift b/Sources/MCSMStream/MCSMStreamClient.swift new file mode 100644 index 0000000..2364de1 --- /dev/null +++ b/Sources/MCSMStream/MCSMStreamClient.swift @@ -0,0 +1,262 @@ +// +// MCSMStreamClient.swift +// MCSMStream +// +// 面板控制台的**实时输出流**客户端（Engine.IO v4 + Socket.IO v5 的最小子集，走 +// `URLSessionWebSocketTask`，不引第三方依赖）。 +// +// 为什么自己写：面板只有 JS 的 `socket.io-client`；iOS/macOS 侧没有官方 Swift 包， +// 而我们需要的只有"连上 → 发一句 stream/auth → 收 instance/stdout 文本"这么点东西。 +// 协议本身十来行，自己实现比引一个没人维护的封装更可控（且能单测）。 +// +// 取消/重连的语义： +// - 口令是**一次性的**，所以每次重连都要重新申请通道 → 客户端拿的是一个 `makeEndpoint` +// 闭包，而不是一个固定 URL； +// - 重试用指数退避，全部失败后**不报错崩溃**，而是发 `.fallback`，由调用方退回轮询 +// （面板 `dataPlane=direct` 且节点在 NAT 后面时，直连本来就不可达，这是正常情况）。 +// + +import Foundation +import MCSMAPI + +public actor MCSMStreamClient { + + // MARK: - 对外事件 + + public enum Event: Sendable, Equatable { + /// 第 n 次尝试（1 起）：正在申请通道 / 建立连接。 + case connecting(attempt: Int) + /// 鉴权通过，`stream/auth` 回了 `true` —— 从现在起可以收实时输出。 + case ready + /// 一段终端输出（原样文本，可能不带换行；交给 `ConsoleTextParser` 处理）。 + case output(String) + /// 实例生命周期：`opened` / `stopped` / `failure`。 + case lifecycle(String) + /// 放弃实时流（附原因），调用方应回退轮询。 + case fallback(reason: String) + } + + public struct Configuration: Sendable { + /// 等 `stream/auth` 回来的上限；超时按本次失败处理（口令一次性，重连会重新申请）。 + public var handshakeTimeout: Duration + /// 最多尝试几次（含首次）。 + public var maxAttempts: Int + /// 首次退避间隔，之后每次翻倍。 + public var backoff: Duration + + public init(handshakeTimeout: Duration = .seconds(15), + maxAttempts: Int = 3, + backoff: Duration = .milliseconds(600)) { + self.handshakeTimeout = handshakeTimeout + self.maxAttempts = maxAttempts + self.backoff = backoff + } + } + + public enum Failure: Error, Equatable, Sendable, CustomStringConvertible { + case timeout + case rejected + case closed + case badResponse(String) + + public var description: String { + switch self { + case .timeout: return "等待面板响应超时" + case .rejected: return "面板拒绝了这个口令" + case .closed: return "连接被关闭" + case let .badResponse(text): return "面板返回了意外内容：\(text.prefix(80))" + } + } + } + + // MARK: - 状态 + + private let makeEndpoint: @Sendable () async throws -> MCSMStreamEndpoint + private let session: URLSession + private let configuration: Configuration + + private var task: URLSessionWebSocketTask? + private var isReady = false + private var stopped = false + + public init( + configuration: Configuration = Configuration(), + session: URLSession = URLSession(configuration: .ephemeral), + makeEndpoint: @escaping @Sendable () async throws -> MCSMStreamEndpoint + ) { + self.configuration = configuration + self.session = session + self.makeEndpoint = makeEndpoint + } + + // MARK: - 生命周期 + + /// 连接并持续产出事件流。返回的流结束 = 已放弃实时流（最后一条是 `.fallback`）。 + public func events() -> AsyncStream { + AsyncStream { continuation in + let runner = Task { [weak self] in + guard let self else { continuation.finish(); return } + await self.run(continuation) + continuation.finish() + } + continuation.onTermination = { _ in + runner.cancel() + Task { await self.close() } + } + } + } + + /// 主动关闭（界面退出 / 切走时调用）。 + public func close() { + stopped = true + if let task { + // 先打个招呼再断，服务端能立刻回收订阅。 + task.send(.string(SocketIOFrame.disconnectPacket)) { _ in } + task.cancel(with: .goingAway, reason: nil) + } + task = nil + isReady = false + } + + /// 通过实时流发一条控制台命令（面板会当成 `instance.execPreset("command", …)`）。 + /// 没连上时抛 `Failure.closed`，调用方回退 HTTP 接口。 + public func sendCommand(_ command: String) async throws { + guard let task, isReady else { throw Failure.closed } + let payload = SocketIOFrame.event("stream/input", data: ["command": command]) + try await task.send(.string(payload)) + } + + // MARK: - 主循环 + + private func run(_ continuation: AsyncStream.Continuation) async { + var attempt = 0 + var delay = configuration.backoff + var lastReason = "" + while !stopped, attempt < maxAttemptsSafe { + attempt += 1 + continuation.yield(.connecting(attempt: attempt)) + do { + try await connectOnce(continuation) + // `connectOnce` 正常返回 = 服务端主动结束（实例停了 / 面板重启）， + // 不当作失败，直接结束（界面会靠状态刷新得知）。 + return + } catch { + if stopped || Task.isCancelled { return } + lastReason = (error as? CustomStringConvertible)?.description ?? "\(error)" + } + if attempt < maxAttemptsSafe { + try? await Task.sleep(for: delay) + delay = delay * 2 + } + } + guard !stopped else { return } + continuation.yield(.fallback(reason: lastReason.isEmpty ? "实时流不可用" : lastReason)) + } + + private var maxAttemptsSafe: Int { max(1, configuration.maxAttempts) } + + /// 一次完整连接：申请通道 → WS 握手 → `stream/auth` → 转发输出，直到断开。 + private func connectOnce(_ continuation: AsyncStream.Continuation) async throws { + let endpoint = try await makeEndpoint() + + let socket = session.webSocketTask(with: endpoint.url) + task = socket + isReady = false + socket.resume() + defer { + socket.cancel(with: .goingAway, reason: nil) + if task === socket { task = nil } + isReady = false + } + + var handshakeDeadline: Task? + + while !stopped { + let message = try await receive(on: socket, timeout: handshakeTimeoutIfNeeded()) + handshakeDeadline?.cancel() + handshakeDeadline = nil + + switch message { + case let .string(text): + let frame = SocketIOFrame.decode(text) + switch frame { + case .open: + try await socket.send(.string(SocketIOFrame.connectPacket)) + case .connected: + let auth = SocketIOFrame.event("stream/auth", data: ["password": endpoint.password]) + try await socket.send(.string(auth)) + // 鉴权回来之前先挂个超时，避免"连上了但永远没 auth"。 + let timeout = configuration.handshakeTimeout + handshakeDeadline = Task { [weak self] in + try? await Task.sleep(for: timeout) + guard let self, await self.isNotReady else { return } + await self.cancelSocket(socket) + } + case .ping: + try await socket.send(.string(SocketIOFrame.pongPacket)) + case .close, .disconnected: + if isReady { return } + throw Failure.closed + case let .connectError(detail): + throw Failure.badResponse(detail) + case let .event(name, _): + if case let accepted? = frame.authAccepted { + guard accepted else { throw Failure.rejected } + isReady = true + handshakeDeadline?.cancel() + handshakeDeadline = nil + continuation.yield(.ready) + // 顺便要一次实例详情（面板会把状态/配置推回来，用于顶部状态行）。 + try await socket.send(.string(SocketIOFrame.event("stream/detail", data: [:]))) + continue + } + if let text = frame.stdoutText { + continuation.yield(.output(text)) + continue + } + if let lifecycle = frame.instanceLifecycle { + continuation.yield(.lifecycle(lifecycle)) + continue + } + _ = name + case .pong, .unknown: + break + } + case .data: + // 面板只发文本；二进制帧直接忽略。 + break + @unknown default: + break + } + } + handshakeDeadline?.cancel() + } + + private var isNotReady: Bool { !isReady } + + private func cancelSocket(_ socket: URLSessionWebSocketTask) { + socket.cancel(with: .goingAway, reason: nil) + } + + /// 握手阶段的每次 `receive` 都带超时；连上之后无限等（服务端会自己 ping）。 + private func handshakeTimeoutIfNeeded() -> Duration? { + isReady ? nil : configuration.handshakeTimeout + } + + private func receive( + on socket: URLSessionWebSocketTask, + timeout: Duration? + ) async throws -> URLSessionWebSocketTask.Message { + guard let timeout else { return try await socket.receive() } + return try await withThrowingTaskGroup(of: URLSessionWebSocketTask.Message.self) { group in + group.addTask { try await socket.receive() } + group.addTask { + try await Task.sleep(for: timeout) + throw Failure.timeout + } + defer { group.cancelAll() } + guard let message = try await group.next() else { throw Failure.timeout } + return message + } + } +} diff --git a/Sources/MCSMStream/SocketIOFrame.swift b/Sources/MCSMStream/SocketIOFrame.swift new file mode 100644 index 0000000..8e6a9b1 --- /dev/null +++ b/Sources/MCSMStream/SocketIOFrame.swift @@ -0,0 +1,177 @@ +// +// SocketIOFrame.swift +// MCSMStream +// +// Engine.IO v4 + Socket.IO v5 的**最小帧编解码**（只覆盖面板终端用到的那几种）。 +// +// 线上就是一条条 WebSocket 文本帧，首字符是 Engine.IO 包类型： +// +// | 前缀 | 含义 | 方向 | +// |---|---|---| +// | `0` | open（带 `{sid, pingInterval, pingTimeout}`） | 服务端 → 客户端 | +// | `1` | close | 双向 | +// | `2` | ping | 服务端 → 客户端（v4 由服务端发，客户端回 `3`） | +// | `3` | pong | 客户端 → 服务端 | +// | `40` | socket.io connect（默认命名空间） | 双向 | +// | `41` | disconnect | 双向 | +// | `42` | event，后面是 `["事件名", 参数…]` | 双向 | +// | `44` | connect_error | 服务端 → 客户端 | +// +// 面板终端用到的业务事件（参数统一是 `{ "data": … }` 外壳）： +// +// - 客户端发：`stream/auth {data:{password}}`、`stream/detail {}`、`stream/input {data:{command}}`； +// - 服务端发：`stream/auth {data:true|false}`、`instance/stdout {data:{instanceUuid,text}}`、 +// `stream/detail {data:{…}}`、`instance/stopped` / `instance/opened` / `instance/failure`。 +// +// 实现取舍：事件参数**保留原始 JSON 文本**（`dataJSON`）而不是 `[String: Any]` —— +// `Any` 既不是 `Equatable` 也不是 `Sendable`，存进枚举会让整个类型没法比较/跨并发域传递； +// 需要取值时用下面几个便利属性按需解析（一次解析，成本可忽略）。 +// + +import Foundation + +/// 一条解析后的 Engine.IO / Socket.IO 帧。 +public enum SocketIOFrame: Equatable, Sendable { + + /// `0`：握手成功，带服务端的 `sid` 等参数（原文 JSON，本包不解析）。 + case open(String) + /// `1`：服务端要求关闭。 + case close + /// `2`：服务端 ping（必须回 `3`）。 + case ping + /// `3`：服务端回了我们的 pong（本项目不发 ping，保留解析以便诊断）。 + case pong + /// `40`：socket.io 命名空间已连接。 + case connected + /// `41`：socket.io 断开。 + case disconnected + /// `42`：业务事件（`name` + 原始 JSON 参数）。 + case event(name: String, dataJSON: String) + /// `44`：连接被拒绝（如鉴权失败）。 + case connectError(String) + /// 其它（二进制帧、未知包）—— 记录但不处理，别让一条怪帧把连接搞挂。 + case unknown(String) + + /// 解析一条**文本**帧。 + public static func decode(_ text: String) -> SocketIOFrame { + guard let first = text.first else { return .unknown(text) } + switch first { + case "0": + return .open(String(text.dropFirst())) + case "1": + return .close + case "2": + return .ping + case "3": + return .pong + case "4": + // `40` / `41` / `42` / `44` + let rest = text.dropFirst() + guard let kind = rest.first else { return .unknown(text) } + switch kind { + case "0": + return .connected + case "1": + return .disconnected + case "2": + return Self.decodeEvent(String(rest.dropFirst())) + case "4": + return .connectError(String(rest.dropFirst())) + default: + return .unknown(text) + } + default: + return .unknown(text) + } + } + + /// `42["stream/auth",{"data":true}]` → `.event(name:dataJSON:)` + static func decodeEvent(_ json: String) -> SocketIOFrame { + guard let data = json.data(using: .utf8), + let array = try? JSONSerialization.jsonObject(with: data) as? [Any], + let name = array.first as? String + else { return .unknown(json) } + guard array.count > 1 else { return .event(name: name, dataJSON: "null") } + // 规范化成排序键的紧凑 JSON：同样的参数总是得到同样的字符串（可断言、可比较）。 + let dataJSON: String + if let encoded = try? JSONSerialization.data( + withJSONObject: array[1], options: [.sortedKeys, .fragmentsAllowed, .withoutEscapingSlashes] + ), let text = String(data: encoded, encoding: .utf8) { + dataJSON = text + } else { + dataJSON = "null" + } + return .event(name: name, dataJSON: dataJSON) + } + + // MARK: - 编码 + + /// 建立默认命名空间连接（握手收到 `0` 之后发）。 + public static let connectPacket = "40" + /// 回 ping。 + public static let pongPacket = "3" + /// 主动断开 socket.io 命名空间。 + public static let disconnectPacket = "41" + + /// 发一个业务事件：`42["name",{"data":…}]`。 + public static func event(_ name: String, data: Any?) -> String { + let payload: [String: Any] = ["data": data ?? NSNull()] + // `.withoutEscapingSlashes`：`stream/auth` 里的 `/` 不必转义成 `\/`， + // 面板/节点两种都认，但不转义时可读性好得多（日志、测试断言都直接看）。 + guard let encoded = try? JSONSerialization.data( + withJSONObject: [name, payload], + options: [.withoutEscapingSlashes] + ), let json = String(data: encoded, encoding: .utf8) + else { return "42[\"\(name)\",{}]" } + return "42" + json + } +} + +// MARK: - 事件参数取值 + +extension SocketIOFrame { + + /// `stream/auth` 的 `data == true`？不是这个事件时 `nil`。 + public var authAccepted: Bool? { + guard case let .event(name, dataJSON) = self, name == "stream/auth" else { return nil } + return Self.boolValue(forKey: "data", in: dataJSON) + } + + /// `instance/stdout` 的文本（不是该事件 / 没有文本时 `nil`）。 + public var stdoutText: String? { + guard case let .event(name, dataJSON) = self, name == "instance/stdout" else { return nil } + guard let data = Self.object(forKey: "data", in: dataJSON) else { return nil } + return data["text"] as? String + } + + /// 实例生命周期事件 → `opened` / `stopped` / `failure`。 + public var instanceLifecycle: String? { + guard case let .event(name, _) = self else { return nil } + switch name { + case "instance/opened": return "opened" + case "instance/stopped": return "stopped" + case "instance/failure": return "failure" + default: return nil + } + } + + /// 事件参数里的某个键（`{"data": …}` 的外壳由调用方自己指定键名）。 + public func value(forKey key: String) -> Any? { + guard case let .event(_, dataJSON) = self else { return nil } + return Self.object(forKey: key, in: dataJSON) + } + + private static func object(in json: String) -> [String: Any]? { + guard let data = json.data(using: .utf8) else { return nil } + return (try? JSONSerialization.jsonObject(with: data)) as? [String: Any] + } + + private static func object(forKey key: String, in json: String) -> [String: Any]? { + object(in: json)?[key] as? [String: Any] + } + + private static func boolValue(forKey key: String, in json: String) -> Bool? { + guard let root = object(in: json) else { return nil } + return root[key] as? Bool + } +} diff --git a/Sources/MCSMStream/StreamEndpoint.swift b/Sources/MCSMStream/StreamEndpoint.swift new file mode 100644 index 0000000..6653b6c --- /dev/null +++ b/Sources/MCSMStream/StreamEndpoint.swift @@ -0,0 +1,115 @@ +// +// StreamEndpoint.swift +// MCSMStream +// +// 把面板给的通道信息（`addr` + `prefix`）解析成一条可以直连的 socket.io WebSocket URL。 +// +// 规则照抄面板前端 `frontend/src/tools/protocol.ts` 的 `parseForwardAddress`， +// 只保留我们用得到的部分： +// +// - `addr` 带 `ws://` / `wss://` / `http://` / `https://` 时按它定协议，否则跟随面板自己的协议； +// - `127.0.0.1` / `localhost` 开头的地址换成本面板的主机名（节点与面板常同机部署， +// 面板返回的就是节点视角的 `127.0.0.1:24444`，从客户端当然连不上）； +// - socket.io 的路径是 `prefix + "/socket.io"`（`prefix` 只在节点配置了路径前缀时才非空， +// 面板 proxy 模式下为空）。 +// + +import Foundation +import MCSMAPI + +public struct MCSMStreamEndpoint: Equatable, Sendable { + + /// 已经拼好的 WebSocket URL（含 `?EIO=4&transport=websocket`）。 + public let url: URL + /// 一次性口令（`stream/auth` 要发的内容）。 + public let password: String + + public init(url: URL, password: String) { + self.url = url + self.password = password + } +} + +extension MCSMStreamEndpoint { + + public enum Error: Swift.Error, Equatable, Sendable, CustomStringConvertible { + /// `addr` 里没有主机名，或拼出来的 URL 不合法。 + case invalidAddress(String) + + public var description: String { + switch self { + case let .invalidAddress(raw): + return "面板返回的地址无法解析：\(raw)" + } + } + } + + /// 由通道 + 面板地址算出连接目标。 + /// + /// - Parameters: + /// - channel: 面板 `stream_channel` 的返回。 + /// - panelURL: 用户在 app 里填的面板地址（用来决定 `ws`/`wss` 和回环地址的替换）。 + public init(channel: MCSMStreamChannel, panelURL: URL) throws { + let resolved = Self.resolve(address: channel.address, prefix: channel.prefix, panelURL: panelURL) + guard let resolved else { throw Error.invalidAddress(channel.address) } + url = resolved + password = channel.password + } + + /// 解析核心（纯函数，单测直接打这里）。 + static func resolve(address rawAddress: String, prefix rawPrefix: String, panelURL: URL) -> URL? { + var address = rawAddress.trimmingCharacters(in: .whitespacesAndNewlines) + guard !address.isEmpty else { return nil } + + // 1. 协议：显式写了就用写的，否则跟随面板（https → wss）。 + let lower = address.lowercased() + var scheme = panelURL.scheme?.lowercased() == "http" ? "ws" : "wss" + for (candidate, websocket) in [ + ("wss://", "wss"), ("ws://", "ws"), ("https://", "wss"), ("http://", "ws") + ] where lower.hasPrefix(candidate) { + scheme = websocket + address = String(address.dropFirst(candidate.count)) + break + } + + // 2. 拆 host / port / path（IPv6 这里不管：面板地址不会是 `[::1]`）。 + var path = "" + if let slash = address.firstIndex(of: "/") { + path = String(address[slash...]) + address = String(address[.. MCSMStreamEndpoint { + let channel = MCSMStreamChannel(address: "127.0.0.1:\(port)", password: password) + // 回环地址会被换成本面板主机名，这里让面板就是 127.0.0.1，正好保持直连。 + return MCSMStreamEndpoint( + url: MCSMStreamEndpoint.resolve( + address: channel.address, + prefix: "", + panelURL: URL(string: "http://127.0.0.1")! + )!, + password: channel.password + ) + } + + /// 收事件直到满足条件或超时；返回期间收到的所有事件。 + private func collect( + _ stream: AsyncStream, + until condition: @escaping (MCSMStreamClient.Event) -> Bool, + timeout: Duration = .seconds(8) + ) async -> [MCSMStreamClient.Event] { + await withTaskGroup(of: [MCSMStreamClient.Event]?.self) { group in + group.addTask { + var events: [MCSMStreamClient.Event] = [] + for await event in stream { + events.append(event) + if condition(event) { return events } + } + return events + } + group.addTask { + try? await Task.sleep(for: timeout) + return nil + } + let first = await group.next() ?? nil + group.cancelAll() + return first ?? [] + } + } + + @Test("鉴权通过后能收到实时输出") + func receivesOutput() async throws { + let server = try FakeSocketIOServer(script: .init( + expectedPassword: "tok", + stdoutAfterAuth: ["[12:00:00] Done (1.2s)!\r\n", "[12:00:01] [Server thread/INFO]: 世界已保存\r\n"] + )) + defer { server.stop() } + + let client = MCSMStreamClient(configuration: .init(maxAttempts: 1)) { + self.endpoint(server.port) + } + let events = await collect(await client.events()) { event in + if case .output = event { return true } + return false + } + await client.close() + + #expect(events.contains(.ready)) + #expect(events.contains(.output("[12:00:00] Done (1.2s)!\r\n"))) + // 服务端确实收到了 `stream/auth` 与 `stream/detail` + let frames = server.frames + #expect(frames.contains { $0.hasPrefix("42[\"stream/auth\"") }) + #expect(frames.contains { $0.hasPrefix("42[\"stream/detail\"") }) + } + + @Test("口令被拒：重试用尽后落到 .fallback（调用方回退轮询）") + func rejectsWrongPassword() async throws { + let server = try FakeSocketIOServer(script: .init(expectedPassword: "right")) + defer { server.stop() } + + let client = MCSMStreamClient(configuration: .init(maxAttempts: 2, backoff: .milliseconds(50))) { + self.endpoint(server.port, password: "wrong") + } + let events = await collect(await client.events()) { event in + if case .fallback = event { return true } + return false + } + await client.close() + + #expect(events.contains { if case .fallback = $0 { return true } else { return false } }) + #expect(!events.contains(.ready)) + #expect(!events.contains { if case .connecting(attempt: 1) = $0 { return true } else { return false } } == false) + } + + @Test("服务端 ping 要回 pong（否则面板 25 秒后踢连接）") + func repliesToPing() async throws { + let server = try FakeSocketIOServer(script: .init(stdoutAfterAuth: ["hello"], pingAfterOpen: true)) + defer { server.stop() } + + let client = MCSMStreamClient(configuration: .init(maxAttempts: 1)) { + self.endpoint(server.port) + } + _ = await collect(await client.events()) { event in event == .ready } + // 给 pong 一点时间回到服务端 + try await Task.sleep(for: .milliseconds(300)) + await client.close() + #expect(server.frames.contains("3")) + } + + @Test("实例退出事件透传为 .lifecycle") + func forwardsLifecycle() async throws { + let server = try FakeSocketIOServer(script: .init(stdoutAfterAuth: ["bye"], stopAfterAuth: true)) + defer { server.stop() } + + let client = MCSMStreamClient(configuration: .init(maxAttempts: 1)) { + self.endpoint(server.port) + } + let events = await collect(await client.events()) { $0 == .lifecycle("stopped") } + await client.close() + #expect(events.contains(.lifecycle("stopped"))) + } + + @Test("连不上时重试用尽 → .fallback，不崩不卡") + func serverUnreachable() async throws { + // 拿一个刚被释放的端口，连上去必然失败。 + let port = try await { + let server = try FakeSocketIOServer(script: .init()) + let value = server.port + server.stop() + return value + }() + + let client = MCSMStreamClient(configuration: .init(maxAttempts: 2, backoff: .milliseconds(50))) { + self.endpoint(port) + } + let events = await collect(await client.events(), until: { event in + if case .fallback = event { return true } + return false + }, timeout: .seconds(20)) + await client.close() + #expect(events.contains { if case .fallback = $0 { return true } else { return false } }) + } +} diff --git a/Tests/MCSMStreamTests/SocketIOFrameTests.swift b/Tests/MCSMStreamTests/SocketIOFrameTests.swift new file mode 100644 index 0000000..f189b4f --- /dev/null +++ b/Tests/MCSMStreamTests/SocketIOFrameTests.swift @@ -0,0 +1,81 @@ +// +// SocketIOFrameTests.swift +// MCSMStreamTests +// +// 帧编解码：这些字符串都是面板/节点实际会发的形状（Engine.IO v4 + Socket.IO v5）。 +// + +import Testing + +@testable import MCSMStream + +@Suite("Socket.IO 帧编解码") +struct SocketIOFrameTests { + + @Test("握手 / 心跳 / 连接包") + func controlPackets() { + #expect(SocketIOFrame.decode("0{\"sid\":\"abc\",\"pingInterval\":25000}") == .open("{\"sid\":\"abc\",\"pingInterval\":25000}")) + #expect(SocketIOFrame.decode("2") == .ping) + #expect(SocketIOFrame.decode("3") == .pong) + #expect(SocketIOFrame.decode("40") == .connected) + #expect(SocketIOFrame.decode("41") == .disconnected) + #expect(SocketIOFrame.decode("1") == .close) + #expect(SocketIOFrame.decode("44{\"message\":\"unauthorized\"}") == .connectError("{\"message\":\"unauthorized\"}")) + } + + @Test("业务事件：取名字与 data 外壳") + func eventDecoding() { + let auth = SocketIOFrame.decode("42[\"stream/auth\",{\"data\":true}]") + #expect(auth == SocketIOFrame.event(name: "stream/auth", dataJSON: "{\"data\":true}")) + #expect(auth.authAccepted == true) + + let rejected = SocketIOFrame.decode("42[\"stream/auth\",{\"data\":false}]") + #expect(rejected.authAccepted == false) + + let stdout = SocketIOFrame.decode( + "42[\"instance/stdout\",{\"data\":{\"instanceUuid\":\"u1\",\"text\":\"[12:00:00] Done\\r\\n\"}}]" + ) + #expect(stdout.stdoutText == "[12:00:00] Done\r\n") + + let stopped = SocketIOFrame.decode("42[\"instance/stopped\",{\"data\":{\"instanceUuid\":\"u1\"}}]") + #expect(stopped.instanceLifecycle == "stopped") + #expect(SocketIOFrame.decode("42[\"instance/opened\",{}]").instanceLifecycle == "opened") + #expect(SocketIOFrame.decode("42[\"instance/failure\",{}]").instanceLifecycle == "failure") + } + + @Test("不是我们的事件时取不到值") + func unrelatedEvent() { + let frame = SocketIOFrame.decode("42[\"something/else\",{\"data\":true}]") + #expect(frame.authAccepted == nil) + #expect(frame.stdoutText == nil) + #expect(frame.instanceLifecycle == nil) + } + + @Test("怪帧不崩：空串 / 半截 JSON / 二进制前缀 / 没有参数") + func malformedFrames() { + let empty = SocketIOFrame.decode("") + #expect(empty == SocketIOFrame.unknown("")) + + let halfJSON = SocketIOFrame.decode("42[") + #expect(halfJSON == SocketIOFrame.unknown("[")) + + let noArgs = SocketIOFrame.decode(#"42["stream/auth"]"#) + #expect(noArgs == SocketIOFrame.event(name: "stream/auth", dataJSON: "null")) + + let binary = SocketIOFrame.decode(#"451-["binary"]"#) + #expect(binary == SocketIOFrame.unknown(#"451-["binary"]"#)) + + let stranger = SocketIOFrame.decode("9") + #expect(stranger == SocketIOFrame.unknown("9")) + } + + @Test("编码：事件带 data 外壳") + func encoding() { + #expect(SocketIOFrame.event("stream/detail", data: [:]) == "42[\"stream/detail\",{\"data\":{}}]") + #expect( + SocketIOFrame.event("stream/auth", data: ["password": "p1"]) + == "42[\"stream/auth\",{\"data\":{\"password\":\"p1\"}}]" + ) + #expect(SocketIOFrame.event("stream/input", data: ["command": "list"]) == "42[\"stream/input\",{\"data\":{\"command\":\"list\"}}]") + } +} diff --git a/Tests/MCSMStreamTests/StreamEndpointTests.swift b/Tests/MCSMStreamTests/StreamEndpointTests.swift new file mode 100644 index 0000000..a129207 --- /dev/null +++ b/Tests/MCSMStreamTests/StreamEndpointTests.swift @@ -0,0 +1,128 @@ +// +// StreamEndpointTests.swift +// MCSMStreamTests +// +// 地址解析：规则来自面板前端 `parseForwardAddress`（见 StreamEndpoint.swift 的注释）。 +// + +import Foundation +import Testing + +@testable import MCSMAPI +@testable import MCSMStream + +@Suite("通道地址解析") +struct StreamEndpointTests { + + private func url(_ address: String, prefix: String = "", panel: String = "https://mcs.example.com") -> String? { + MCSMStreamEndpoint.resolve( + address: address, + prefix: prefix, + panelURL: URL(string: panel)! + )?.absoluteString + } + + @Test("proxy 模式：面板地址 + 空前缀 + https → wss") + func proxyMode() { + #expect(url("mcs.example.com") == "wss://mcs.example.com/socket.io?EIO=4&transport=websocket") + } + + @Test("direct 模式：节点给了端口") + func directWithPort() { + #expect( + url("10.0.0.5:24444") == "wss://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + ) + } + + @Test("回环地址换成本面板的主机名（保留端口）") + func loopbackBecomesPanelHost() { + #expect( + url("127.0.0.1:24444") == "wss://mcs.example.com:24444/socket.io?EIO=4&transport=websocket" + ) + #expect( + url("localhost:24444") == "wss://mcs.example.com:24444/socket.io?EIO=4&transport=websocket" + ) + #expect( + url("127.0.0.1") == "wss://mcs.example.com/socket.io?EIO=4&transport=websocket" + ) + } + + @Test("节点配了路径前缀（两侧斜杠都容忍）") + func prefixPath() { + #expect( + url("10.0.0.5:24444", prefix: "/daemon/") + == "wss://10.0.0.5:24444/daemon/socket.io?EIO=4&transport=websocket" + ) + #expect( + url("10.0.0.5:24444", prefix: "daemon") + == "wss://10.0.0.5:24444/daemon/socket.io?EIO=4&transport=websocket" + ) + } + + @Test("地址自带协议时以它为准") + func explicitScheme() { + #expect( + url("ws://10.0.0.5:24444") == "ws://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + ) + #expect( + url("http://10.0.0.5:24444") == "ws://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + ) + #expect( + url("https://10.0.0.5:24444") == "wss://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + ) + } + + @Test("面板走 http 时用 ws") + func insecurePanel() { + #expect( + url("mcs.example.com:81", panel: "http://mcs.example.com:81") + == "ws://mcs.example.com:81/socket.io?EIO=4&transport=websocket" + ) + } + + @Test("空地址 / 空主机 → 解析失败") + func invalidAddress() { + #expect(url("") == nil) + #expect(url(" ") == nil) + #expect(MCSMStreamEndpoint.resolve(address: "", prefix: "", panelURL: URL(string: "https://x")!) == nil) + } + + @Test("通道模型：带外壳与裸形状都能解") + func channelFromResponse() throws { + let payload = Components.Schemas.StreamChannelPayload(addr: "127.0.0.1:24444", password: "p1") + let wrapped = Components.Schemas.StreamChannelResponse(status: 200, data: payload) + let channel = try #require(MCSMStreamChannel(wrapped)) + #expect(channel.address == "127.0.0.1:24444") + #expect(channel.password == "p1") + #expect(channel.prefix == "") + + let bare = Components.Schemas.StreamChannelResponse(status: nil, addr: "a.example.com", password: "p2", prefix: "/node") + #expect(MCSMStreamChannel(bare)?.prefix == "/node") + + let empty = Components.Schemas.StreamChannelResponse(status: 200) + #expect(MCSMStreamChannel(empty) == nil) + } + + @Test("通道模型：从 text/plain 的 JSON 文本也能解") + func channelFromText() throws { + let text = """ + {"status":200,"data":{"addr":"127.0.0.1:24444","password":"tok","prefix":"","remoteMappings":[]},"time":1} + """ + let channel = try #require(MCSMStreamChannel(text: text)) + #expect(channel.address == "127.0.0.1:24444") + #expect(channel.password == "tok") + #expect(channel.remoteMappings.isEmpty) + + #expect(MCSMStreamChannel(text: "{\"status\":500,\"data\":\"节点未连接\"}") == nil) + #expect(MCSMStreamChannel(text: "not json") == nil) + } + + @Test("端点的 URL 与口令打包") + func endpoint() throws { + let channel = MCSMStreamChannel(address: "mcs.example.com", password: "tok") + let endpoint = try MCSMStreamEndpoint(channel: channel, panelURL: URL(string: "https://mcs.example.com")!) + #expect(endpoint.password == "tok") + #expect(endpoint.url.scheme == "wss") + #expect(endpoint.url.path == "/socket.io") + } +} From 4e7df302f0457d3b2ef0a675e413bc02652f1c46 Mon Sep 17 00:00:00 2001 From: wangzhizhou <824219521@qq.com> Date: Tue, 27 Oct 2026 08:06:47 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(stream):=20socket.io=20=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E5=BF=85=E9=A1=BB=E4=BB=A5=20`/`=20=E7=BB=93=E5=B0=BE?= =?UTF-8?q?=20+=20=E7=9C=9F=E9=9D=A2=E6=9D=BF=E5=86=92=E7=83=9F=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **真机实测踩到的坑**：`wss://节点:443/socket.io?EIO=4&transport=websocket`（少尾部斜杠） 被反向代理直接 404，`URLSessionWebSocketTask` 表现为 `-1011 Websocket handshake failure`（看不到 404 的原因）。面板前端的 socket.io 客户端请求的是 `/socket.io/?EIO=4…`，所以路径必须补上尾部 `/`。 - `StreamEndpoint.resolve`：`/socket.io` → `/socket.io/`（并把这行注释写清楚，别再改回去）； - 单测的期望 URL 同步更新； - 新增 `LiveStreamTests`（**默认跳过**，靠 `ORZMC_MCSM_LIVE_*` 环境变量启用）：真面板上 申请通道 → 连 socket.io → 期望 `stream/auth` 回 `true` → 发 `/list` → 期望收到实时输出。 等待是**有界**的（实例安静时实时流本来就不发东西，不能无限等）。 实测（`mcs.jokerhub.cn`，direct 模式，节点 `wss://mcs-node.jokerhub.cn:443`）： ``` [live] addr=wss://mcs-node.jokerhub.cn:443 prefix= mappings=0 [live] ready=true firstOutput=/l\u{1b}[31m/li\u{1b}[0m\u{1b}[K… ``` —— 鉴权通过，输出带 ANSI 码，说明协议、路径、口令投递都对。 --- README.md | 47 ++++++++ Sources/MCSMStream/StreamEndpoint.swift | 5 +- Tests/MCSMStreamTests/LiveStreamTests.swift | 108 ++++++++++++++++++ .../MCSMStreamTests/StreamEndpointTests.swift | 22 ++-- 4 files changed, 170 insertions(+), 12 deletions(-) create mode 100644 Tests/MCSMStreamTests/LiveStreamTests.swift diff --git a/README.md b/README.md index 9276eb9..fc4a92d 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,38 @@ let text = try MCSMResponseEnvelope.checkedText(raw) // 或 unwrapText(raw) | `GET /api/protected_instance/restart` | 同上 | 重启 | | `GET /api/protected_instance/command` | 同上 | 命令走查询参数；`/` 开头是 MC 命令，不带 `/` 是控制台命令 | | `GET /api/protected_instance/outputlog` | 同上 | 返回 `text/plain` 的**日志尾部**，可能从半行开始。体里仍是 `{status,data,time}` 外壳 → 用 `MCSMResponseEnvelope` 解 | +| `POST /api/protected_instance/stream_channel` | 同上 | 控制台**实时流**的第一步：拿 `addr` + 一次性口令，然后连那台 socket.io 发 `stream/auth`。见下节 | + +### 实时流（`MCSMStream`，另一个 product） + +轮询 `outputlog` 的实时性差；面板支持把节点 stdout 一条条推过来（面板自己的前端就是这么做的）： + +```swift +import MCSMStream + +let channel = try await MCSMStreamChannel(client.getStreamChannel(query: .init(daemonId: d, uuid: u)))! +let endpoint = try MCSMStreamEndpoint(channel: channel, panelURL: panelURL) + +let stream = MCSMStreamClient { endpoint } +for await event in await stream.events() { + switch event { + case .ready: break // stream/auth 通过，可以收输出了 + case .output(let text): buffer.append(text) + case .fallback(let reason): await poll() // 实时流不可用 → 回退轮询 + case .connecting, .lifecycle: break + } +} +``` + +要点（都是踩过的）： + +- **一次性口令**：每次重连都要重新申请通道 → 客户端收的是 `makeEndpoint` 闭包，不是固定 URL； +- **`addr` 可能是节点视角**（`127.0.0.1:24444`，面板 `dataPlane=direct` 时就是节点的公开地址）： + 回环地址会被换成本面板的主机名；连不上（节点在 NAT 后面）就回退轮询，这是正常路径； +- **`/socket.io/` 的尾部斜杠不能少**：少了会 404 / 握手失败（实测 `-1011 Websocket handshake failure`）； +- **必须回 pong**：Engine.IO v4 由服务端 ping，不回会被 20~25 秒后踢掉； +- 事件参数形如 `{ "data": … }`，`instance/stdout` 的 `data.text` 是**原始终端文本**（带 ANSI 码、 + 可能半行、可能 `\r` 覆盖）→ 交给 `ConsoleTextParser` 那一套渲染，别自己按行切。 ### 响应外壳（重要） @@ -129,6 +161,21 @@ let text = try MCSMResponseEnvelope.checkedText(raw) // 或 unwrapText(raw) 它会拉官方 markdown，逐个检查 spec 里声明的路径还在不在（CI 里也有一个**不阻塞**的 job 跑它）。 +## 真实面板冒烟（可选） + +`stream_channel` 的形状与 socket.io 可达性只有真面板能验证，所以有一个默认跳过的测试： + +```bash +ORZMC_MCSM_LIVE_URL=https://mcs.example.com \ +ORZMC_MCSM_LIVE_KEY= \ +ORZMC_MCSM_LIVE_DAEMON= \ +ORZMC_MCSM_LIVE_INSTANCE= \ +swift test --filter LiveStreamTests +``` + +它会：申请通道 → 连 socket.io → 期望 `stream/auth` 回 `true` → 发一条 `/list` → 期望收到实时输出。 +口令与实例**不要**写进仓库。 + ## 测试 ```bash diff --git a/Sources/MCSMStream/StreamEndpoint.swift b/Sources/MCSMStream/StreamEndpoint.swift index 6653b6c..cf59e55 100644 --- a/Sources/MCSMStream/StreamEndpoint.swift +++ b/Sources/MCSMStream/StreamEndpoint.swift @@ -97,7 +97,10 @@ extension MCSMStreamEndpoint { let prefix = rawPrefix.trimmingCharacters(in: .whitespacesAndNewlines) .trimmingCharacters(in: CharacterSet(charactersIn: "/")) var socketPath = prefix.isEmpty ? "/socket.io" : "/\(prefix)/socket.io" - if socketPath.hasSuffix("/") { socketPath.removeLast() } + // ⚠️ 必须以 `/` 结尾：socket.io 客户端请求的是 `/socket.io/?EIO=4…`。 + // 少了这个斜杠，节点（真机实测：`mcs-node.*:443` 反向代理）直接 404/握手失败 —— + // 我之前就是这么踩的：`-1011 Websocket handshake failure`，而 `/socket.io/` 一切正常。 + if !socketPath.hasSuffix("/") { socketPath += "/" } var components = URLComponents() components.scheme = scheme diff --git a/Tests/MCSMStreamTests/LiveStreamTests.swift b/Tests/MCSMStreamTests/LiveStreamTests.swift new file mode 100644 index 0000000..88d69b7 --- /dev/null +++ b/Tests/MCSMStreamTests/LiveStreamTests.swift @@ -0,0 +1,108 @@ +// +// LiveStreamTests.swift +// MCSMStreamTests +// +// **真实面板**的冒烟测试：默认跳过，只在给了环境变量时跑。用来回答两个只有真机能回答的问题： +// +// 1. `stream_channel` 在这个面板版本上返回什么形状（外壳 / 裸、`direct` 还是 `proxy` 模式）； +// 2. 节点/面板的 socket.io 到底连不连得上、`stream/auth` 会不会回 `true`。 +// +// 用法（口令与实例由调用方给，**不要写进仓库**）： +// +// ``` +// ORZMC_MCSM_LIVE_URL=https://mcs.example.com \ +// ORZMC_MCSM_LIVE_KEY=<面板 API Key> \ +// ORZMC_MCSM_LIVE_DAEMON= \ +// ORZMC_MCSM_LIVE_INSTANCE= \ +// swift test --filter LiveStreamTests +// ``` +// +// 预期：拿得到通道 → 连上 → `.ready`（实例在跑时还能收到 `instance/stdout`）。 +// + +import Foundation +import MCSMAPI +import Testing + +@testable import MCSMStream + +private struct LiveConfiguration { + let panelURL: URL + let apiKey: String + let daemonID: String + let instanceUUID: String + + static var fromEnvironment: LiveConfiguration? { + let env = ProcessInfo.processInfo.environment + guard let url = env["ORZMC_MCSM_LIVE_URL"].flatMap(URL.init(string:)), + let key = env["ORZMC_MCSM_LIVE_KEY"], !key.isEmpty, + let daemon = env["ORZMC_MCSM_LIVE_DAEMON"], !daemon.isEmpty, + let instance = env["ORZMC_MCSM_LIVE_INSTANCE"], !instance.isEmpty + else { return nil } + return LiveConfiguration(panelURL: url, apiKey: key, daemonID: daemon, instanceUUID: instance) + } +} + +@Suite("真实面板冒烟（需环境变量）", .serialized) +struct LiveStreamTests { + + @Test("申请通道 → 连 socket.io → stream/auth 通过", .enabled(if: LiveConfiguration.fromEnvironment != nil)) + func liveChannelAuthenticates() async throws { + let config = try #require(LiveConfiguration.fromEnvironment) + let client = MCSMClient.make(serverURL: config.panelURL, apiKey: config.apiKey) + + let output = try await client.getStreamChannel( + query: .init(daemonId: config.daemonID, uuid: config.instanceUUID) + ) + guard case .ok(let ok) = output else { + Issue.record("stream_channel 未返回 200：\(output)") + return + } + let channel = try #require(MCSMStreamChannel(ok.body.json), "面板没给通道信息") + print("[live] addr=\(channel.address) prefix=\(channel.prefix) mappings=\(channel.remoteMappings.count)") + + let endpoint = try MCSMStreamEndpoint(channel: channel, panelURL: config.panelURL) + print("[live] socket.io=\(endpoint.url.absoluteString)") + + let streamClient = MCSMStreamClient( + configuration: .init(handshakeTimeout: .seconds(20), maxAttempts: 1) + ) { + endpoint + } + var sawReady = false + var firstOutput: String? + + // 有界等待：实时流在"实例在跑但没有新输出"时会一直安静，不能无限等事件。 + let events = await streamClient.events() + await withTaskGroup(of: Void.self) { group in + group.addTask { + for await event in events { + switch event { + case .ready: + sawReady = true + // 连上之后立刻发一条**只读**命令：既验证双向通道（`stream/input`）， + // 也顺带制造一段输出，免得"实例安静"时测试永远等不到 stdout。 + try? await streamClient.sendCommand("/list") + case let .output(text): + if firstOutput == nil { firstOutput = text } + case let .fallback(reason): + print("[live] fallback=\(reason)") + Issue.record("实时流不可用：\(reason)") + case .connecting, .lifecycle: + break + } + if sawReady && firstOutput != nil { return } + } + } + group.addTask { + try? await Task.sleep(for: .seconds(40)) + } + await group.next() + group.cancelAll() + } + await streamClient.close() + print("[live] ready=\(sawReady) firstOutput=\(firstOutput?.prefix(120) ?? "(空)")") + #expect(sawReady, "stream/auth 没有通过") + #expect(firstOutput != nil, "发了 /list 却没收到任何实时输出") + } +} diff --git a/Tests/MCSMStreamTests/StreamEndpointTests.swift b/Tests/MCSMStreamTests/StreamEndpointTests.swift index a129207..af6c249 100644 --- a/Tests/MCSMStreamTests/StreamEndpointTests.swift +++ b/Tests/MCSMStreamTests/StreamEndpointTests.swift @@ -24,26 +24,26 @@ struct StreamEndpointTests { @Test("proxy 模式：面板地址 + 空前缀 + https → wss") func proxyMode() { - #expect(url("mcs.example.com") == "wss://mcs.example.com/socket.io?EIO=4&transport=websocket") + #expect(url("mcs.example.com") == "wss://mcs.example.com/socket.io/?EIO=4&transport=websocket") } @Test("direct 模式：节点给了端口") func directWithPort() { #expect( - url("10.0.0.5:24444") == "wss://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + url("10.0.0.5:24444") == "wss://10.0.0.5:24444/socket.io/?EIO=4&transport=websocket" ) } @Test("回环地址换成本面板的主机名（保留端口）") func loopbackBecomesPanelHost() { #expect( - url("127.0.0.1:24444") == "wss://mcs.example.com:24444/socket.io?EIO=4&transport=websocket" + url("127.0.0.1:24444") == "wss://mcs.example.com:24444/socket.io/?EIO=4&transport=websocket" ) #expect( - url("localhost:24444") == "wss://mcs.example.com:24444/socket.io?EIO=4&transport=websocket" + url("localhost:24444") == "wss://mcs.example.com:24444/socket.io/?EIO=4&transport=websocket" ) #expect( - url("127.0.0.1") == "wss://mcs.example.com/socket.io?EIO=4&transport=websocket" + url("127.0.0.1") == "wss://mcs.example.com/socket.io/?EIO=4&transport=websocket" ) } @@ -51,24 +51,24 @@ struct StreamEndpointTests { func prefixPath() { #expect( url("10.0.0.5:24444", prefix: "/daemon/") - == "wss://10.0.0.5:24444/daemon/socket.io?EIO=4&transport=websocket" + == "wss://10.0.0.5:24444/daemon/socket.io/?EIO=4&transport=websocket" ) #expect( url("10.0.0.5:24444", prefix: "daemon") - == "wss://10.0.0.5:24444/daemon/socket.io?EIO=4&transport=websocket" + == "wss://10.0.0.5:24444/daemon/socket.io/?EIO=4&transport=websocket" ) } @Test("地址自带协议时以它为准") func explicitScheme() { #expect( - url("ws://10.0.0.5:24444") == "ws://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + url("ws://10.0.0.5:24444") == "ws://10.0.0.5:24444/socket.io/?EIO=4&transport=websocket" ) #expect( - url("http://10.0.0.5:24444") == "ws://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + url("http://10.0.0.5:24444") == "ws://10.0.0.5:24444/socket.io/?EIO=4&transport=websocket" ) #expect( - url("https://10.0.0.5:24444") == "wss://10.0.0.5:24444/socket.io?EIO=4&transport=websocket" + url("https://10.0.0.5:24444") == "wss://10.0.0.5:24444/socket.io/?EIO=4&transport=websocket" ) } @@ -76,7 +76,7 @@ struct StreamEndpointTests { func insecurePanel() { #expect( url("mcs.example.com:81", panel: "http://mcs.example.com:81") - == "ws://mcs.example.com:81/socket.io?EIO=4&transport=websocket" + == "ws://mcs.example.com:81/socket.io/?EIO=4&transport=websocket" ) } From 8396aa3cb91926193217388c486c7d02fe620795 Mon Sep 17 00:00:00 2001 From: wangzhizhou <824219521@qq.com> Date: Tue, 27 Oct 2026 08:12:22 +0800 Subject: [PATCH 3/3] =?UTF-8?q?chore(ci):=20apidoc-drift=20=E6=94=BE?= =?UTF-8?q?=E8=BF=87=E5=AE=98=E6=96=B9=E6=96=87=E6=A1=A3=E6=9C=AA=E6=94=B6?= =?UTF-8?q?=E5=BD=95=E7=9A=84=20/api/auth?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `/api/auth` 是面板前端的登录态聚合接口，官方 apidoc 只写了 admin 那套 —— 我们用它列 普通用户的实例（实测 10.18.0 可用，README「权限」节有记录）。之前这个 job 一直红着， 红久了就没人看它了，反而放过真正的 drift。现在加**显式白名单**（带理由），其余端点照旧严查： 新加的 `/api/protected_instance/stream_channel` 就在官方文档里，能正常校验到。 （顺带记一笔：这条提交信息第一版用 `-m "…\`…\`…"` 写的，反引号被 shell 当命令执行掉了 —— AGENTS.md §7 早就写了"多行提交信息一律写进文件再 `-F`"，这次又踩了一次。） --- scripts/check-apidoc-drift.sh | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/scripts/check-apidoc-drift.sh b/scripts/check-apidoc-drift.sh index 900a48e..b3f90db 100755 --- a/scripts/check-apidoc-drift.sh +++ b/scripts/check-apidoc-drift.sh @@ -18,11 +18,24 @@ for f in api_instance.md api_daemon.md; do curl -fsSL "${BASE}/${f}" >>"${docs}" done +# 官方文档里**确实没有**、但我们已确认在用的端点（面板内部路由）。 +# 加进这里必须写清楚证据，别拿它掩盖真正的 drift。 +undocumented_ok() { + case "$1" in + # 面板 `/api/auth` 是前端登录态的聚合接口，官方 apidoc 只写了 admin 那套。 + # 我们用它列"普通用户自己的实例"，实测 10.18.0 可正常返回（见 README「权限」节）。 + "/api/auth") return 0 ;; + *) return 1 ;; + esac +} + status=0 # 从 spec 里取出所有路径（/api/...），逐个在官方文档里找。 while IFS= read -r path; do if grep -Fq "${path}" "${docs}"; then echo "[ok] ${path}" + elif undocumented_ok "${path}"; then + echo "[skip] ${path}（官方文档未收录，已确认在用）" else echo "[drift] 官方 apidoc 里找不到 ${path}（端点可能改名/下架，需同步 spec）" >&2 status=1