diff --git a/README.md b/README.md index 72b36bd..019004a 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,7 @@ Backend、Agent、Probe 以 OCI 镜像发布(Linux amd64、arm64),Agent | 文档 | 内容 | | --- | --- | | [`docs/architecture.md`](docs/architecture.md) | 架构、数据链路、安全边界、设计决策 | +| [`docs/security.md`](docs/security.md) | 零信任设计稿(尚未实现):威胁模型、名册和操作员凭证、监听隐身、extension 授权 | | [`docs/protocol.md`](docs/protocol.md)、[`backend/fleet/README.md`](backend/fleet/README.md) | 线协议、Fleet 节点框架(身份、注册、连接) | | [`docs/enrollment.md`](docs/enrollment.md) | 注册、身份、Probe 与 Agent 的接入 | | [`docs/measurement.md`](docs/measurement.md)、[`docs/analysis.md`](docs/analysis.md) | 采集引擎与任务模型、分析层与指标 | diff --git a/TODO.md b/TODO.md index 9bf0d20..7d45c7c 100644 --- a/TODO.md +++ b/TODO.md @@ -12,12 +12,12 @@ ## 任务与面板 - 单个 Agent 的任务(目前是线路模板加按 Agent 的覆盖,没有只属于某一个 Agent 的独立任务) -- 原生 webhook 投递;告警(基于 webhook) +- 原生 webhook 投递;告警(基于 webhook)。实现之前先定约束(只允许 HTTPS、禁止私网和回环、不跟随重定向、限制超时和响应大小、解析后再检查地址),见 [`docs/security.md`](docs/security.md) 第 9.2 节 ## 部署 - 后端自动下载前端资源:用 Podman、Docker 部署时,默认拉取最新的前端资源(现在前端要自己构建,镜像里没有)。 - +[security.md](docs/security.md) ## 访客页面 - 人机验证:访客页面要通过验证才能拉数据,前端访问时静默触发。候选 tiago2/cap 和 Cloudflare Turnstile。 @@ -38,7 +38,7 @@ ## 安全 -- **Agent、Backend、Probe 的零信任设计**:目前 Backend 是三者之间的信任中心,需要梳理哪些信任可以去掉。 +- **零信任**:设计稿在 [`docs/security.md`](docs/security.md),按其中第 12 节的顺序实现(阶段零「协议方向声明表」已完成):Probe 自己生成密钥、Agent 预握手敲门、操作员凭证和名册、测量数据的签名和归属校验、入库配额和管理端 CSP、Backend 应用层身份、extension 框架(敏感能力都放进可选安装的独立 extension,核心保持 NativeAOT);可选的 UDP 单包授权、UDP 会话迁移的路径验证。 - **extension**(Web SSH、TTY、更新遥测,敏感 operator):设计时必须保住「节点是纯采集器」这条边界,见 [`README.md`](README.md) 的项目边界。 - 对 UDP 包、控制帧、RawChunk、JSON-RPC 做模糊测试;伪造签名和重放的专项测试。 diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index 26e4435..50ae208 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -16,7 +16,8 @@ public sealed class AgentApp : IDisposable private readonly AgentConfig _config; private readonly bool _ipv6Supported; - private readonly RestartController? _restart; + private readonly RestartController _restart; + private readonly RestartController? _ownedRestart; private readonly LoadedNodeIdentity _identity; private readonly TimeProvider _time; private readonly INodeLog _log; @@ -40,7 +41,8 @@ public sealed class AgentApp : IDisposable /// 配置目录(容器里的 /etc/moshiro):宿主机更新器和节点在这里交换清单、请求和状态。 public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log, RestartController? restart = null) { - _restart = restart; + _ownedRestart = restart is null ? new RestartController() : null; + _restart = restart ?? _ownedRestart!; _update = new UpdateBridge(dataDirectory, new ConfigStore(dataDirectory).Write, log); _config = config; _identity = identity; @@ -102,7 +104,11 @@ public sealed class AgentApp : IDisposable /// 数据面和控制端口都开始监听之后完成(配置里的端口为 0 时,这里能读到实际端口)。 public Task Ready => _ready.Task; - public void Dispose() => _udp.Dispose(); + public void Dispose() + { + _udp.Dispose(); + _ownedRestart?.Dispose(); + } internal StreamRegistry Streams => _streams; @@ -116,6 +122,9 @@ public sealed class AgentApp : IDisposable public FleetLink Link => _link; + /// 对等面(Probe 连进来的会话)的路由器,测试用。 + public RpcRouter PeerRouter => _hub.Router; + /// 期望状态里是否已经有这个 Probe 的公钥(测试用:握手会按它校验)。 public bool IsPeerKnown(string probeId, string publicKeyBase64) => _tasks.LookupPeerKey(probeId) is { } key && Convert.ToBase64String(key) == publicKeyBase64; @@ -190,7 +199,7 @@ public sealed class AgentApp : IDisposable private FleetLink BuildLink() { - var router = new RpcRouter(); + var router = new RpcRouter(RpcSide.Node, MoShiroMethodTable.Table, NodeRoles.Agent); HashSet? allowed = _config.Policy.AllowMethods is { } list ? [.. list] : null; bool Allow(string method) => allowed is null || allowed.Contains(method); @@ -221,7 +230,7 @@ public sealed class AgentApp : IDisposable }); } - if (Allow(MoShiroMethods.Schedule) && _restart is not null) + if (Allow(MoShiroMethods.Schedule)) { router.OnNotify(MoShiroMethods.Schedule, MoShiroJsonContext.Default.ScheduleParams, (_, schedule, _) => { diff --git a/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs b/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs index 77acdd2..5a4a1da 100644 --- a/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs +++ b/backend/agent/src/MoShiro.Agent/Peer/PeerHub.cs @@ -90,6 +90,9 @@ public sealed class PeerHub private readonly TimeProvider _time; private readonly INodeLog _log; + /// 对等面入站方法的路由器(只读,测试用)。 + public RpcRouter Router => _router; + public PeerHub(SessionRegistry sessions, MonoClock clock, ChunkAssembler chunks, TaskManager tasks, BackendUplink uplink, TimeProvider time, INodeLog log) { _router = BuildRouter(); @@ -272,7 +275,7 @@ public sealed class PeerHub } private static RpcRouter BuildRouter() => - new RpcRouter() + new RpcRouter(RpcSide.Listener, MoShiroMethodTable.Table) .On( MoShiroMethods.Clock, MoShiroJsonContext.Default.ClockParams, diff --git a/backend/fleet/README.md b/backend/fleet/README.md index fea7018..ae5ffdd 100644 --- a/backend/fleet/README.md +++ b/backend/fleet/README.md @@ -56,6 +56,7 @@ - JSON-RPC 2.0;方法名 `领域.动作`,`fleet.*` 保留给框架,产品用自己的前缀(MoShiro 是 `moshiro.*`)。 - 二进制帧 `u8 channel | 正文`:`0`–`15` 框架保留,产品从 `16` 开始。 - 错误:`error.data.code` 为 snake_case,框架的以 `fleet_` 开头,产品可以扩展。 +- **方法必须声明**:每个方法在声明表(`MethodTable`)里写明发起方、请求还是通知、哪些角色可以参与;不存在双向方法,同一个方法名在一个平面里只有一个发起方。`RpcRouter` 在注册、分发和发送三处核对,违规回 `fleet_protocol_violation`,超过上限断开连接。详见 [`docs/security.md`](../../docs/security.md) 第 8 节。 - 与传输无关:同一套路由跑在 wss 上,也跑在 TLS 帧(`u32 长度 | u8 类型 | 正文`)上。 | 保留方法 | 方向 | 说明 | @@ -85,11 +86,13 @@ connect(wss) → fleet.challenge → fleet.auth → fleet.hello → 状态同步 ```text 节点 A(发起方,如 Probe)── TLS(按 SPKI 校验证书)──> 节点 B(监听方,如 Agent) - <── fleet.challenge ── - ── fleet.auth(peer 用途,A 的私钥签名)──> + <── fleet.peer.challenge ── + ── fleet.peer.auth(peer 用途,A 的私钥签名)──> B 用服务端下发的对端公钥列表验签(经 state.sync 下发),之后复用同一套 Rpc:路由、二进制通道、发件箱 ``` +握手用两个方法:`fleet.peer.challenge`(监听方→拨号方,通知)和 `fleet.peer.auth`(拨号方→监听方,请求);握手完成后,两侧各自只能发起声明表里属于自己这一侧的方法。 + 对端公钥和允许连接的节点由产品通过期望状态下发给监听方,Fleet 只管认证和会话,拓扑由产品决定;UDP、原始 TCP 数据面由产品自己实现,Fleet 只提供会话密钥派生和首帧 MAC 的辅助函数。 ## 宿主与服务端抽象 diff --git a/backend/fleet/src/Fleet.Core/ErrorCode.cs b/backend/fleet/src/Fleet.Core/ErrorCode.cs index 8f2cf03..b80e184 100644 --- a/backend/fleet/src/Fleet.Core/ErrorCode.cs +++ b/backend/fleet/src/Fleet.Core/ErrorCode.cs @@ -13,6 +13,8 @@ public static class FleetErrors public static readonly ErrorCode Unauthorized = new("fleet_unauthorized"); public static readonly ErrorCode Forbidden = new("fleet_forbidden"); public static readonly ErrorCode Unsupported = new("fleet_unsupported"); + /// 方法没有声明、方向不对、类型不对或角色不允许:对方违反了协议。 + public static readonly ErrorCode ProtocolViolation = new("fleet_protocol_violation"); public static readonly ErrorCode InvalidParams = new("fleet_invalid_params"); public static readonly ErrorCode NotFound = new("fleet_not_found"); public static readonly ErrorCode Internal = new("fleet_internal"); diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs index 5d86ded..04a19d1 100644 --- a/backend/fleet/src/Fleet.Link/FleetLink.cs +++ b/backend/fleet/src/Fleet.Link/FleetLink.cs @@ -105,6 +105,9 @@ public sealed class FleetLink public RpcSession? CurrentSession => _session; + /// 入站方法的路由器(只读,测试用它核对声明表和实际注册的处理函数)。 + public RpcRouter Router => _router; + /// 被服务端撤销或禁用:不会再重连。 public bool Revoked => _stopped; diff --git a/backend/fleet/src/Fleet.Rpc/FleetMethods.cs b/backend/fleet/src/Fleet.Rpc/FleetMethods.cs index 6062ead..d189253 100644 --- a/backend/fleet/src/Fleet.Rpc/FleetMethods.cs +++ b/backend/fleet/src/Fleet.Rpc/FleetMethods.cs @@ -15,6 +15,31 @@ public static class FleetMethods public const string Revoked = "fleet.revoked"; } +/// +/// 框架自己的方法声明,两个平面各一份。认证和握手阶段由连接代码直接处理的(Inline)不经过路由器。 +/// 同一个方法名在两个平面里可以有不同的发起方:fleet.state.sync 在节点面是服务端发给节点,在对等面是 Agent 发给 Probe。 +/// +public static class FleetMethodTable +{ + public static readonly MethodTable Table = new( + [ + // 节点面 + MethodSpec.Notification(FleetMethods.Challenge, RpcSide.Server).AsInline(), + MethodSpec.Request(FleetMethods.Auth, RpcSide.Node).AsInline(), + MethodSpec.Request(FleetMethods.Hello, RpcSide.Node).AsInline(), + MethodSpec.Notification(FleetMethods.Heartbeat, RpcSide.Node).AsInline(), + MethodSpec.Request(FleetMethods.StateSync, RpcSide.Server), + MethodSpec.Notification(FleetMethods.OutboxAck, RpcSide.Server), + MethodSpec.Notification(FleetMethods.Revoked, RpcSide.Server), + + // 对等面 + MethodSpec.Notification(PeerMethods.Challenge, RpcSide.Listener).AsInline(), + MethodSpec.Request(PeerMethods.Auth, RpcSide.Dialer).AsInline(), + MethodSpec.Request(FleetMethods.StateSync, RpcSide.Listener), + MethodSpec.Notification(FleetMethods.OutboxAck, RpcSide.Listener), + ]); +} + /// 二进制通道:0–15 保留给框架,产品从 16 开始。 public static class FleetChannels { diff --git a/backend/fleet/src/Fleet.Rpc/MethodTable.cs b/backend/fleet/src/Fleet.Rpc/MethodTable.cs new file mode 100644 index 0000000..71b9641 --- /dev/null +++ b/backend/fleet/src/Fleet.Rpc/MethodTable.cs @@ -0,0 +1,120 @@ +namespace Fleet.Rpc; + +/// 连接的哪一侧。节点面:服务端和节点;对等面:拨号方(Probe)和监听方(Agent)。 +public enum RpcSide +{ + Server, + Node, + Dialer, + Listener, +} + +public enum RpcPlane +{ + Node, + Peer, +} + +public enum MethodKind +{ + Request, + Notification, +} + +public static class RpcSides +{ + public static RpcPlane PlaneOf(RpcSide side) => side is RpcSide.Server or RpcSide.Node ? RpcPlane.Node : RpcPlane.Peer; + + public static RpcSide Opposite(RpcSide side) => side switch + { + RpcSide.Server => RpcSide.Node, + RpcSide.Node => RpcSide.Server, + RpcSide.Dialer => RpcSide.Listener, + _ => RpcSide.Dialer, + }; +} + +/// +/// 一个方法的声明:只有 能发起它(不存在双向方法),类型是请求还是通知,哪些节点角色可以参与。 +/// 表示它在认证或握手阶段由连接代码直接处理,不经过路由器。 +/// +public sealed record MethodSpec(string Name, RpcSide From, MethodKind Kind, IReadOnlyList? Roles = null, bool Inline = false) +{ + public RpcPlane Plane => RpcSides.PlaneOf(From); + + /// 接收方。 + public RpcSide To => RpcSides.Opposite(From); + + public bool AllowsRole(string? role) => Roles is null || (role is not null && Roles.Contains(role)); + + public static MethodSpec Request(string name, RpcSide from, params string[] roles) => new(name, from, MethodKind.Request, roles.Length == 0 ? null : roles); + + public static MethodSpec Notification(string name, RpcSide from, params string[] roles) => new(name, from, MethodKind.Notification, roles.Length == 0 ? null : roles); + + public MethodSpec AsInline() => this with { Inline = true }; +} + +/// 二进制通道的声明:谁能发、哪些角色可以发。通道 0–15 保留给框架,产品从 16 开始。 +public sealed record ChannelSpec(byte Channel, RpcSide From, IReadOnlyList? Roles = null) +{ + public RpcPlane Plane => RpcSides.PlaneOf(From); + + public bool AllowsRole(string? role) => Roles is null || (role is not null && Roles.Contains(role)); +} + +/// +/// 协议方法和二进制通道的声明表,单一真源。路由器用它在注册时拒绝方向、类型、角色不符的处理函数, +/// 在收发时拒绝没有声明或不该由对方发起的方法。同一个方法名在不同平面里可以有不同的发起方,所以按(平面,名字)查找。 +/// +public sealed class MethodTable +{ + private readonly Dictionary<(RpcPlane, string), MethodSpec> _methods = []; + private readonly Dictionary<(RpcPlane, byte), ChannelSpec> _channels = []; + + public MethodTable(IEnumerable methods, IEnumerable? channels = null) + { + foreach (MethodSpec spec in methods) + { + if (!_methods.TryAdd((spec.Plane, spec.Name), spec)) + { + throw new ArgumentException($"method {spec.Name} is declared twice in the {spec.Plane} plane"); + } + } + + foreach (ChannelSpec spec in channels ?? []) + { + if (!_channels.TryAdd((spec.Plane, spec.Channel), spec)) + { + throw new ArgumentException($"channel {spec.Channel} is declared twice in the {spec.Plane} plane"); + } + } + } + + public IEnumerable Methods => _methods.Values; + + public IEnumerable Channels => _channels.Values; + + public MethodSpec? Find(RpcPlane plane, string method) => _methods.GetValueOrDefault((plane, method)); + + public ChannelSpec? FindChannel(RpcPlane plane, byte channel) => _channels.GetValueOrDefault((plane, channel)); + + /// 合并多张表(框架的加产品的)。 + public static MethodTable Combine(params MethodTable[] tables) => + new(tables.SelectMany(t => t.Methods), tables.SelectMany(t => t.Channels)); + + /// + /// 对账: 这一侧、 这个角色应该处理、但没有注册处理函数的方法 + /// (认证阶段直接处理的 方法除外)。 为空表示不限角色(服务端要处理所有角色发来的)。 + /// + public IReadOnlyList MissingHandlers(RpcSide receiver, string? role, IEnumerable registered) + { + HashSet have = [.. registered]; + return + [ + .. Methods + .Where(m => m.To == receiver && !m.Inline && (role is null || m.AllowsRole(role)) && !have.Contains(m.Name)) + .Select(m => m.Name) + .Order(StringComparer.Ordinal), + ]; + } +} diff --git a/backend/fleet/src/Fleet.Rpc/RpcRouter.cs b/backend/fleet/src/Fleet.Rpc/RpcRouter.cs index eb362ac..ba4ffe3 100644 --- a/backend/fleet/src/Fleet.Rpc/RpcRouter.cs +++ b/backend/fleet/src/Fleet.Rpc/RpcRouter.cs @@ -4,7 +4,17 @@ using Fleet.Core; namespace Fleet.Rpc; -/// 方法路由,与传输方式无关。 由传输层提供(例如调用方节点)。 +/// 对端违反了协议:方法没有声明、方向不对、请求和通知用反了、或者角色不允许。 +public readonly record struct RpcViolation(string Method, string Reason); + +/// 一次分发的结果:要回的响应(通知为空),以及对端有没有违规。 +public readonly record struct RpcDispatch(RpcMessage? Response, RpcViolation? Violation); + +/// +/// 方法路由,与传输方式无关,也是协议方向的唯一卡口:节点和服务端、Probe 和 Agent 的入站分发都经过这里。 +/// 由传输层提供(例如调用方节点)。路由器知道自己在连接的哪一侧()和声明表, +/// 注册处理函数时核对方向、类型和角色,分发时拒绝没有声明或不该由对方发起的方法,发送时核对自己有没有资格发起。 +/// public sealed class RpcRouter { private delegate ValueTask> Route(TContext context, JsonElement? parameters, CancellationToken ct); @@ -13,12 +23,28 @@ public sealed class RpcRouter private readonly Dictionary _routes = new(StringComparer.Ordinal); private readonly Action? _onUnhandled; + private readonly string? _localRole; + private readonly Func? _callerRole; - public RpcRouter(Action? onUnhandled = null) + /// 本路由器在连接的哪一侧。 + /// 方法声明表。 + /// 本节点的角色(节点侧和对等面用):声明限定了角色的方法,角色不符就不能注册。 + /// 从调用上下文取出对端的角色(服务端用):声明限定了角色的方法,角色不符就拒绝。 + public RpcRouter(RpcSide side, MethodTable table, string? localRole = null, Func? callerRole = null, Action? onUnhandled = null) { + Side = side; + Table = table; + _localRole = localRole; + _callerRole = callerRole; _onUnhandled = onUnhandled; } + public RpcSide Side { get; } + + public MethodTable Table { get; } + + private RpcPlane Plane => RpcSides.PlaneOf(Side); + public bool Has(string method) => _routes.ContainsKey(method); public IReadOnlyCollection Methods => _routes.Keys; @@ -29,6 +55,7 @@ public sealed class RpcRouter JsonTypeInfo resultInfo, Func>> handler) { + Declare(method, MethodKind.Request); _routes[method] = async (context, raw, ct) => { if (!TryDeserialize(raw, paramsInfo, out TParams? p)) @@ -49,6 +76,7 @@ public sealed class RpcRouter JsonTypeInfo paramsInfo, Func handler) { + Declare(method, MethodKind.Notification); _routes[method] = async (context, raw, ct) => { if (!TryDeserialize(raw, paramsInfo, out TParams? p)) @@ -66,12 +94,26 @@ public sealed class RpcRouter /// 处理一条请求或通知。请求返回响应;通知返回 null(失败时只丢弃)。 /// 处理函数里的异常属于程序缺陷:在这个边界捕获,回复 internal 错误,并交给 onUnhandled 记录。 /// - public async ValueTask DispatchAsync(TContext context, RpcMessage message, CancellationToken ct) + public async ValueTask DispatchAsync(TContext context, RpcMessage message, CancellationToken ct) => + (await DispatchCheckedAsync(context, message, ct)).Response; + + /// + /// 先核对声明(方向、类型、角色),再分发。违规的消息不会到达处理函数,返回的结果里带 , + /// 由会话决定怎么办(计数、超过上限就断开连接)。声明过、方向对、但这一侧没有处理函数的方法不算违规,只是不支持。 + /// + public async ValueTask DispatchCheckedAsync(TContext context, RpcMessage message, CancellationToken ct) { string method = message.Method ?? ""; + if (Check(context, message) is { } violation) + { + return new RpcDispatch( + message.Id is null ? null : RpcMessage.Failure(message.Id, new FleetError(FleetErrors.ProtocolViolation, violation.Reason)), + violation); + } + if (!_routes.TryGetValue(method, out Route? route)) { - return message.Id is null ? null : RpcMessage.Failure(message.Id, new FleetError(FleetErrors.Unsupported, $"unknown method {method}")); + return new RpcDispatch(message.Id is null ? null : RpcMessage.Failure(message.Id, new FleetError(FleetErrors.Unsupported, $"unknown method {method}")), null); } Result result; @@ -91,12 +133,90 @@ public sealed class RpcRouter if (message.Id is null) { - return null; + return new RpcDispatch(null, null); } - return result.TryGet(out JsonElement? value, out FleetError? error) - ? new RpcMessage { Id = message.Id, Result = value ?? s_emptyObject } - : RpcMessage.Failure(message.Id, error); + return new RpcDispatch( + result.TryGet(out JsonElement? value, out FleetError? error) + ? new RpcMessage { Id = message.Id, Result = value ?? s_emptyObject } + : RpcMessage.Failure(message.Id, error), + null); + } + + /// 入站检查:这条消息是不是对方有资格发起的、类型对不对、对方的角色允不允许。 + public RpcViolation? Check(TContext context, RpcMessage message) + { + string method = message.Method ?? ""; + MethodSpec? spec = Table.Find(Plane, method); + if (spec is null) + { + return new RpcViolation(method, $"method {method} is not declared"); + } + + if (spec.To != Side) + { + return new RpcViolation(method, $"{method} cannot be initiated by the {RpcSides.Opposite(Side)} side"); + } + + MethodKind kind = message.Id is null ? MethodKind.Notification : MethodKind.Request; + if (spec.Kind != kind) + { + return new RpcViolation(method, $"{method} is a {spec.Kind.ToString().ToLowerInvariant()}"); + } + + if (_callerRole is not null && !spec.AllowsRole(_callerRole(context))) + { + return new RpcViolation(method, $"{method} is not allowed for this node role"); + } + + return null; + } + + /// + /// 发送前检查:本侧有没有资格发起这个方法、类型对不对、对端的角色允不允许(服务端发给节点时传入节点的角色)。 + /// 违反是程序缺陷,抛 。 + /// + public void CheckSend(string method, MethodKind kind, string? peerRole = null) + { + MethodSpec? spec = Table.Find(Plane, method); + if (spec is null) + { + throw new InvalidOperationException($"method {method} is not declared"); + } + + if (spec.From != Side) + { + throw new InvalidOperationException($"{method} cannot be initiated by the {Side} side"); + } + + if (spec.Kind != kind) + { + throw new InvalidOperationException($"{method} is a {spec.Kind.ToString().ToLowerInvariant()}"); + } + + if (peerRole is not null && !spec.AllowsRole(peerRole)) + { + throw new InvalidOperationException($"{method} cannot be sent to a {peerRole}"); + } + } + + private void Declare(string method, MethodKind kind) + { + MethodSpec spec = Table.Find(Plane, method) ?? throw new InvalidOperationException($"method {method} is not declared in the method table"); + if (spec.To != Side) + { + throw new InvalidOperationException($"{method} is initiated by the {spec.From} side and cannot be handled on the {Side} side"); + } + + if (spec.Kind != kind) + { + throw new InvalidOperationException($"{method} is declared as a {spec.Kind.ToString().ToLowerInvariant()}"); + } + + if (_localRole is not null && !spec.AllowsRole(_localRole)) + { + throw new InvalidOperationException($"{method} is not for the {_localRole} role"); + } } private static bool TryDeserialize(JsonElement? raw, JsonTypeInfo info, [System.Diagnostics.CodeAnalysis.NotNullWhen(true)] out T? value) diff --git a/backend/fleet/src/Fleet.Rpc/RpcSession.cs b/backend/fleet/src/Fleet.Rpc/RpcSession.cs index 8ca127c..ec1770a 100644 --- a/backend/fleet/src/Fleet.Rpc/RpcSession.cs +++ b/backend/fleet/src/Fleet.Rpc/RpcSession.cs @@ -34,6 +34,9 @@ public sealed class RpcSessionOptions /// 待处理的入站请求上限;超过时接收循环等待(背压)。 public int InboundQueue { get; set; } = 256; + + /// 对端违反协议(方法没有声明、方向、类型或角色不对)的次数上限;超过就断开会话。 + public int MaxViolations { get; set; } = 20; } /// @@ -53,6 +56,8 @@ public sealed class RpcSession : IAsyncDisposable private readonly CancellationTokenSource _stop = new(); private long _nextId; private long _lastReceive; + private int _violations; + private bool _violated; public RpcSession(IRpcTransport transport, RpcRouter router, TContext context, TimeProvider time, RpcSessionOptions? options = null) { @@ -81,7 +86,7 @@ public sealed class RpcSession : IAsyncDisposable while (!linked.IsCancellationRequested) { RpcFrame? frame = await _transport.ReceiveAsync(linked.Token); - if (frame is null) + if (frame is null || Volatile.Read(ref _violated)) { return; } @@ -184,8 +189,16 @@ public sealed class RpcSession : IAsyncDisposable public ValueTask NotifyAsync(string method, TParams parameters, JsonTypeInfo paramsInfo, CancellationToken ct) => SendAsync(RpcMessage.Notification(method, parameters, paramsInfo), ct); - public ValueTask SendAsync(RpcMessage message, CancellationToken ct) => - _transport.SendAsync(new RpcFrame(FrameKind.Text, JsonSerializer.SerializeToUtf8Bytes(message, FleetRpcContext.Default.RpcMessage)), ct); + public ValueTask SendAsync(RpcMessage message, CancellationToken ct) + { + if (message.Method is not null) + { + // 请求和通知只能发本侧有资格发起的方法(响应没有方法名,不检查) + _router.CheckSend(message.Method, message.Id is null ? MethodKind.Notification : MethodKind.Request); + } + + return _transport.SendAsync(new RpcFrame(FrameKind.Text, JsonSerializer.SerializeToUtf8Bytes(message, FleetRpcContext.Default.RpcMessage)), ct); + } public ValueTask SendBinaryAsync(ReadOnlyMemory frame, CancellationToken ct) => _transport.SendAsync(new RpcFrame(FrameKind.Binary, frame), ct); @@ -205,10 +218,18 @@ public sealed class RpcSession : IAsyncDisposable { await foreach (RpcMessage message in _inbound.Reader.ReadAllAsync(ct)) { - RpcMessage? response = await _router.DispatchAsync(_context, message, ct); - if (response is not null) + RpcDispatch outcome = await _router.DispatchCheckedAsync(_context, message, ct); + if (outcome.Response is not null) { - await SendAsync(response, ct); + await SendAsync(outcome.Response, ct); + } + + if (outcome.Violation is not null && ++_violations > _options.MaxViolations) + { + // 只关闭传输,不取消接收:接收循环在下一次读到帧或对端确认关闭时正常结束 + Volatile.Write(ref _violated, true); + await _transport.CloseAsync("protocol violations", CancellationToken.None); + return; } } } diff --git a/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs b/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs index 0a581bf..2020147 100644 --- a/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs +++ b/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs @@ -12,4 +12,7 @@ public sealed class FleetServerOptions public TimeSpan HeartbeatTimeout { get; set; } = TimeSpan.FromSeconds(45); public TimeSpan RequestTimeout { get; set; } = TimeSpan.FromSeconds(15); + + /// 节点违反协议(方法没有声明、方向、类型或角色不对)的次数上限;超过就断开这条连接。 + public int MaxViolations { get; set; } = 20; } diff --git a/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs b/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs index d2bf663..b06fa57 100644 --- a/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs +++ b/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs @@ -33,6 +33,7 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier Inboxes = inboxes.ToDictionary(i => i.Channel); Sinks = [.. sinks]; Router = router.Inner; + RoleOf = router.RoleOf; } internal IFleetStore Store { get; } @@ -51,6 +52,8 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier internal RpcRouter Router { get; } + internal Func? RoleOf { get; } + public IReadOnlyCollection OnlineNodes => [.. _connections.Keys]; public bool IsOnline(NodeId node) => _connections.ContainsKey(node); @@ -161,10 +164,18 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier /// 产品注册的 RPC 方法路由。 public sealed class FleetRouter { - public FleetRouter(RpcRouter inner) + public FleetRouter(RpcRouter inner, Func? roleOf) { Inner = inner; + RoleOf = roleOf; } internal RpcRouter Inner { get; } + + internal Func? RoleOf { get; } + + /// 已注册的入站方法名(只读,测试用它核对声明表)。 + public IReadOnlyCollection Methods => Inner.Methods; + + public MethodTable Table => Inner.Table; } diff --git a/backend/fleet/src/Fleet.Server.Link/FleetLog.cs b/backend/fleet/src/Fleet.Server.Link/FleetLog.cs index 3c80dd2..10963c9 100644 --- a/backend/fleet/src/Fleet.Server.Link/FleetLog.cs +++ b/backend/fleet/src/Fleet.Server.Link/FleetLog.cs @@ -19,6 +19,9 @@ internal static partial class FleetLog [LoggerMessage(Level = LogLevel.Warning, Message = "state push failed node={Node} type={Type} err={Err}")] public static partial void StatePushFailed(ILogger logger, string node, string type, string err); + [LoggerMessage(Level = LogLevel.Warning, Message = "protocol violation node={Node} method={Method} reason={Reason}")] + public static partial void ProtocolViolation(ILogger logger, string node, string method, string reason); + [LoggerMessage(Level = LogLevel.Error, Message = "unhandled error in handler method={Method}")] public static partial void HandlerFailed(ILogger logger, Exception exception, string method); diff --git a/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs b/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs index 755deb2..ede5c05 100644 --- a/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs +++ b/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs @@ -12,6 +12,8 @@ namespace Fleet.Server; public sealed class FleetServerBuilder { private readonly List, IServiceProvider>> _routeConfigurators = []; + private MethodTable? _methods; + private Func? _roleOf; internal FleetServerBuilder(IServiceCollection services) { @@ -22,6 +24,21 @@ public sealed class FleetServerBuilder internal IReadOnlyList, IServiceProvider>> RouteConfigurators => _routeConfigurators; + internal MethodTable Methods => _methods ?? throw new InvalidOperationException("call WithMethods(...) when adding the Fleet server: the method table is required"); + + internal Func? RoleOf => _roleOf; + + /// + /// 方法声明表(框架的加产品的,见 )和节点的角色解析(例如按节点 ID 前缀区分 Agent 和 Probe)。必须设置: + /// 路由器据此核对每个处理函数的方向、类型和角色,并拒绝没有声明或不该由节点发起的方法。 + /// + public FleetServerBuilder WithMethods(MethodTable methods, Func roleOf) + { + _methods = methods; + _roleOf = roleOf; + return this; + } + public FleetServerBuilder AddState<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TProvider>() where TProvider : class, IStateProvider { @@ -71,13 +88,18 @@ public static class FleetServerExtensions services.AddSingleton(provider => { ILogger logger = provider.GetRequiredService().CreateLogger("Fleet.Rpc"); - var router = new RpcRouter((method, ex) => FleetLog.HandlerFailed(logger, ex, method)); + Func? roleOf = builder.RoleOf; + var router = new RpcRouter( + RpcSide.Server, + builder.Methods, + callerRole: roleOf is null ? null : context => roleOf(context.Node), + onUnhandled: (method, ex) => FleetLog.HandlerFailed(logger, ex, method)); foreach (Action, IServiceProvider> configurator in builder.RouteConfigurators) { configurator(router, provider); } - return new FleetRouter(router); + return new FleetRouter(router, roleOf); }); services.AddSingleton(provider => new FleetLinkHost( provider.GetRequiredService(), diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Auth.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Auth.cs new file mode 100644 index 0000000..55f33fe --- /dev/null +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Auth.cs @@ -0,0 +1,98 @@ +using System.Net.WebSockets; +using System.Security.Cryptography; +using System.Text.Json; +using Fleet.Core; +using Fleet.Identity; +using Fleet.Rpc; + +namespace Fleet.Server; + +/// 认证:挑战、签名校验、拒绝。 +internal sealed partial class NodeConnection +{ + private async Task AuthenticateAsync(CancellationToken ct) + { + string nonce = Convert.ToBase64String(RandomNumberGenerator.GetBytes(32)); + long expiresAtMs = _host.Time.GetUtcNow().Add(_host.Options.ChallengeTtl).ToUnixTimeMilliseconds(); + await SendAsync( + RpcMessage.Notification(FleetMethods.Challenge, new ChallengeParams { Nonce = nonce, ExpiresAtMs = expiresAtMs }, FleetRpcContext.Default.ChallengeParams), + ct); + + using var timeout = new CancellationTokenSource(s_authTimeout, _host.Time); + using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); + Frame? frame; + try + { + frame = await ReceiveFrameAsync(linked.Token); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + await RejectAsync(null, "auth timeout", ct); + return false; + } + + if (frame is not { IsText: true } || !TryParse(frame.Data, out RpcMessage? message) || message.Method != FleetMethods.Auth || message.Id is null) + { + await RejectAsync(null, "expected fleet.auth", ct); + return false; + } + + AuthParams? auth = null; + try + { + auth = message.Params?.Deserialize(FleetRpcContext.Default.AuthParams); + } + catch (JsonException) + { + // 按缺失处理 + } + + NodeRecord? node = auth is null ? null : await _host.Store.GetNodeAsync(new NodeId(auth.NodeId), ct); + bool signatureOk = node is not null + && _host.Time.GetUtcNow().ToUnixTimeMilliseconds() <= expiresAtMs + && TryDecode(auth!.Signature, out byte[] signature) + && NodeIdentity.Verify(node.PublicKey, signature, SignaturePurpose.Auth, auth.NodeId, nonce); + if (!signatureOk) + { + // 节点不存在与签名错误返回同样的错误,不泄露节点是否存在。 + await RejectAsync(message.Id, "authentication failed", ct); + return false; + } + + _node = node; + if (node!.Status == NodeStatus.Disabled) + { + await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.NodeDisabled, "node is disabled")), ct); + await RevokeAsync("disabled", ct); + return false; + } + + _session = new NodeSessionInfo(node.Id, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), node.NodeVersion, [], null); + await SendAsync( + RpcMessage.Success(message.Id.Value, new AuthResult { Status = StatusName(node.Status) }, FleetRpcContext.Default.AuthResult), + ct); + return true; + } + + private async Task RejectAsync(long? id, string reason, CancellationToken ct) + { + FleetLog.AuthRejected(_host.Logger, reason); + _closeReason ??= reason; + if (id is not null) + { + await SendAsync(RpcMessage.Failure(id, new FleetError(FleetErrors.Unauthorized, "authentication failed")), ct); + } + + if (_socket.State == WebSocketState.Open) + { + try + { + await _socket.CloseOutputAsync(WebSocketCloseStatus.PolicyViolation, "unauthorized", ct); + } + catch (WebSocketException) + { + // 对端已经断开 + } + } + } +} diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs new file mode 100644 index 0000000..7df1ced --- /dev/null +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs @@ -0,0 +1,240 @@ +using System.Text.Json; +using Fleet.Core; +using Fleet.Rpc; + +namespace Fleet.Server; + +/// 接收:读帧、分发消息、hello、心跳、二进制通道(收件箱)。 +internal sealed partial class NodeConnection +{ + private async Task ReceiveLoopAsync(CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + using var idle = new CancellationTokenSource(_host.Options.HeartbeatTimeout, _host.Time); + using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, idle.Token); + Frame? frame; + try + { + frame = await ReceiveFrameAsync(linked.Token); + } + catch (OperationCanceledException) when (idle.IsCancellationRequested && !ct.IsCancellationRequested) + { + _closeReason ??= "heartbeat timeout"; + return; + } + + if (frame is null) + { + _closeReason ??= "closed by peer"; + return; + } + + if (frame.IsText) + { + if (TryParse(frame.Data, out RpcMessage? message)) + { + await HandleMessageAsync(message, ct); + } + } + else + { + await HandleBinaryAsync(frame.Data, ct); + } + } + } + + private async Task HandleMessageAsync(RpcMessage message, CancellationToken ct) + { + if (message.IsResponse) + { + if (_pending.TryGetValue(message.Id!.Value, out TaskCompletionSource? waiter)) + { + waiter.TrySetResult(message); + } + + return; + } + + switch (message.Method) + { + case FleetMethods.Hello: + await HandleHelloAsync(message, ct); + return; + case FleetMethods.Heartbeat: + await HandleHeartbeatAsync(message, ct); + return; + } + + if (_node!.Status != NodeStatus.Active) + { + if (message.Id is not null) + { + await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.NodePending, "node is not approved yet")), ct); + } + + return; + } + + var context = new NodeCallContext(_node, _session!, _host); + RpcDispatch outcome; + try + { + outcome = await _host.Router.DispatchCheckedAsync(context, message, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + + if (outcome.Response is not null) + { + await SendAsync(outcome.Response, ct); + } + + await NoteViolationAsync(outcome.Violation); + } + + /// 记录节点的违规;超过上限就断开这条连接。 + private async Task NoteViolationAsync(RpcViolation? violation) + { + if (violation is not { } v) + { + return; + } + + FleetLog.ProtocolViolation(_host.Logger, _node!.Id.Value, v.Method, v.Reason); + if (++_violations > _host.Options.MaxViolations) + { + await CloseAsync("protocol violations"); + } + } + + private async Task HandleHelloAsync(RpcMessage message, CancellationToken ct) + { + HelloParams? hello = null; + try + { + hello = message.Params?.Deserialize(FleetRpcContext.Default.HelloParams); + } + catch (JsonException) + { + // 按缺失处理 + } + + if (hello is null) + { + if (message.Id is not null) + { + await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.InvalidParams, "invalid hello")), ct); + } + + return; + } + + long nowMs = _host.Time.GetUtcNow().ToUnixTimeMilliseconds(); + _session = _session! with + { + NodeVersion = hello.NodeVersion, + Capabilities = hello.Capabilities, + HelloMetaJson = hello.Meta?.GetRawText(), + }; + _node = _node! with { NodeVersion = hello.NodeVersion }; + _helloDone = true; + + await UpdatePresenceAsync(online: true, ct); + await FireAsync(NodeEventKind.Hello, ct, hello); + + if (message.Id is not null) + { + await SendAsync( + RpcMessage.Success( + message.Id.Value, + new HelloResult { ServerTimeMs = nowMs, ClockOffsetMs = hello.NowUtcMs - nowMs }, + FleetRpcContext.Default.HelloResult), + ct); + } + + SignalStateChanged(); + } + + private async Task HandleHeartbeatAsync(RpcMessage message, CancellationToken ct) + { + JsonElement? extra = null; + try + { + extra = message.Params?.Deserialize(FleetRpcContext.Default.HeartbeatParams)?.Extra; + } + catch (JsonException) + { + // 忽略格式不正确的附加字段 + } + + long now = _host.Time.GetTimestamp(); + if (_host.Time.GetElapsedTime(Interlocked.Read(ref _lastPresenceTicks), now) >= s_presenceRefresh) + { + await UpdatePresenceAsync(online: true, ct); + } + + await FireAsync(NodeEventKind.Heartbeat, ct, hello: null, extra); + if (message.Id is not null) + { + await SendAsync(new RpcMessage { Id = message.Id, Result = JsonSerializer.SerializeToElement(new EmptyObject(), FleetEmptyObject.Info) }, ct); + } + } + + private async Task HandleBinaryAsync(byte[] data, CancellationToken ct) + { + if (_node!.Status != NodeStatus.Active || !OutboxFrame.TryParse(data, out byte channel, out Guid msgId, out ReadOnlyMemory payload)) + { + return; + } + + // 通道也要声明:节点只能往声明过、并且允许它的角色使用的通道发 + ChannelSpec? spec = _host.Router.Table.FindChannel(RpcPlane.Node, channel); + if (spec is null || spec.From != RpcSide.Node || !spec.AllowsRole(_host.RoleOf?.Invoke(_node))) + { + await NoteViolationAsync(new RpcViolation($"channel {channel}", $"channel {channel} is not declared for this node")); + return; + } + + if (!_host.Inboxes.TryGetValue(channel, out IInbox? inbox)) + { + return; + } + + InboxOutcome outcome; + try + { + outcome = await inbox.HandleAsync(_node, msgId, payload, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + // 没有持久化就不确认,节点会重发。 + FleetLog.InboxFailed(_host.Logger, ex, _node.Id.Value, channel); + return; + } + + if (outcome == InboxOutcome.Rejected) + { + return; + } + + var ack = new OutboxAckParams(); + string id = msgId.ToString(); + if (outcome == InboxOutcome.Duplicate) + { + ack.Duplicates.Add(id); + } + else + { + ack.Ids.Add(id); + } + + await SendAsync(RpcMessage.Notification(FleetMethods.OutboxAck, ack, FleetRpcContext.Default.OutboxAckParams), ct); + } +} diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Send.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Send.cs new file mode 100644 index 0000000..467ca87 --- /dev/null +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Send.cs @@ -0,0 +1,120 @@ +using System.Net.WebSockets; +using System.Text.Json; +using Fleet.Core; +using Fleet.Rpc; + +namespace Fleet.Server; + +/// 发给节点:关闭、吊销、发送一条消息、发起请求并等响应。 +internal sealed partial class NodeConnection +{ + /// + /// 服务端发起的关闭:发送 Close 帧,等对端确认之后接收循环自然结束。 + /// 注意不能直接取消接收的令牌——取消 ReceiveAsync 会中止整个 WebSocket,对端拿不到关闭原因;取消只作为超时兜底。 + /// + public async ValueTask CloseAsync(string reason) + { + _closeReason ??= reason; + try + { + await _sendLock.WaitAsync(CancellationToken.None); + } + catch (ObjectDisposedException) + { + return; + } + + try + { + if (_socket.State == WebSocketState.Open) + { + await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, reason, CancellationToken.None); + } + } + catch (WebSocketException) + { + // 对端已经断开 + } + finally + { + _sendLock.Release(); + } + + _lifetime.CancelAfter(s_closeTimeout); + } + + public async ValueTask RevokeAsync(string reason, CancellationToken ct) + { + await SendAsync(RpcMessage.Notification(FleetMethods.Revoked, new RevokedParams { Reason = reason }, FleetRpcContext.Default.RevokedParams), ct); + await CloseAsync(reason); + } + + public async ValueTask SendAsync(RpcMessage message, CancellationToken ct) + { + if (message.Method is not null) + { + // 服务端只能发自己有资格发起的方法,并且对端的角色要允许(响应没有方法名,不检查) + string? peerRole = _node is null ? null : _host.RoleOf?.Invoke(_node); + _host.Router.CheckSend(message.Method, message.Id is null ? MethodKind.Notification : MethodKind.Request, peerRole); + } + + byte[] bytes = JsonSerializer.SerializeToUtf8Bytes(message, FleetRpcContext.Default.RpcMessage); + try + { + await _sendLock.WaitAsync(ct); + } + catch (ObjectDisposedException) + { + return false; + } + + try + { + if (_socket.State != WebSocketState.Open) + { + return false; + } + + await _socket.SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, ct); + return true; + } + catch (WebSocketException) + { + return false; + } + finally + { + _sendLock.Release(); + } + } + + public async ValueTask> RequestAsync(string method, JsonElement parameters, CancellationToken ct) + { + long id = Interlocked.Increment(ref _nextRequestId); + var waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _pending[id] = waiter; + try + { + var request = new RpcMessage { Id = id, Method = method, Params = parameters }; + if (!await SendAsync(request, ct)) + { + return Result.Fail(FleetErrors.NodeOffline, "node is not connected"); + } + + using var timeout = new CancellationTokenSource(_host.Options.RequestTimeout, _host.Time); + using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); + RpcMessage response = await waiter.Task.WaitAsync(linked.Token); + return response.Error is not null + ? Result.Fail(response.Error.ToFleetError()) + : Result.Ok(response.Result ?? default); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + return Result.Fail(FleetErrors.Timeout, $"{method} timed out"); + } + finally + { + _pending.TryRemove(id, out _); + } + } +} diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs new file mode 100644 index 0000000..dafa0e5 --- /dev/null +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs @@ -0,0 +1,123 @@ +using System.Text.Json; +using Fleet.Core; +using Fleet.Rpc; + +namespace Fleet.Server; + +/// 状态:期望状态推送、在线状态、节点事件。 +internal sealed partial class NodeConnection +{ + public void SignalStateChanged() => _stateSignal.Writer.TryWrite(true); + + + private async Task StatePushLoopAsync(CancellationToken ct) + { + while (await _stateSignal.Reader.WaitToReadAsync(ct)) + { + _stateSignal.Reader.TryRead(out _); + bool complete = await PushStateAsync(ct); + if (!complete) + { + await Task.Delay(s_statePushRetry, _host.Time, ct); + SignalStateChanged(); + } + } + } + + private async Task PushStateAsync(CancellationToken ct) + { + NodeRecord? fresh = await _host.Store.GetNodeAsync(_node!.Id, ct); + if (fresh is null) + { + return true; + } + + _node = fresh with { NodeVersion = _node.NodeVersion ?? fresh.NodeVersion }; + if (fresh.Status == NodeStatus.Disabled) + { + await RevokeAsync("disabled", ct); + return true; + } + + if (fresh.Status != NodeStatus.Active || !_helloDone) + { + // pending 不下发状态;还没有 hello 的连接等 hello 之后再推送。 + return true; + } + + bool complete = true; + foreach (IStateProvider provider in _host.StateProviders) + { + StateDocument? document = await provider.GetAsync(_node, ct); + if (document is null || (_lastPushedVersion.TryGetValue(provider.Type, out long last) && last == document.Version)) + { + continue; + } + + Result raw = await RequestAsync( + FleetMethods.StateSync, + JsonSerializer.SerializeToElement( + new StateSyncParams { Type = provider.Type, Version = document.Version, Doc = document.Doc }, + FleetRpcContext.Default.StateSyncParams), + ct); + if (!raw.TryGet(out JsonElement value, out FleetError? error)) + { + FleetLog.StatePushFailed(_host.Logger, _node.Id.Value, provider.Type, error.Code.Value); + complete = false; + continue; + } + + StateSyncResult? result = null; + try + { + result = value.Deserialize(FleetRpcContext.Default.StateSyncResult); + } + catch (JsonException) + { + // 按缺失处理 + } + + if (result is null) + { + complete = false; + continue; + } + + _lastPushedVersion[provider.Type] = document.Version; + await provider.OnAppliedAsync(_node, document.Version, result, ct); + } + + return complete; + } + + + private async Task UpdatePresenceAsync(bool online, CancellationToken ct) + { + Interlocked.Exchange(ref _lastPresenceTicks, _host.Time.GetTimestamp()); + await _host.Store.UpdatePresenceAsync( + _node!.Id, + new Presence(online, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), _session?.NodeVersion, _session?.Capabilities ?? [], _session?.HelloMetaJson), + ct); + } + + private async Task FireAsync(NodeEventKind kind, CancellationToken ct, HelloParams? hello = null, JsonElement? extra = null) + { + var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra); + foreach (INodeEventSink sink in _host.Sinks) + { + try + { + await sink.OnEventAsync(nodeEvent, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + // 产品的事件处理失败不能影响连接。 + FleetLog.SinkFailed(_host.Logger, ex, kind.ToString(), _node!.Id.Value); + } + } + } +} diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs new file mode 100644 index 0000000..0c16154 --- /dev/null +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs @@ -0,0 +1,85 @@ +using System.Buffers; +using System.Net.WebSockets; +using System.Text.Json; +using Fleet.Rpc; + +namespace Fleet.Server; + +/// 线上的帧:WebSocket 收帧、解析、状态名。 +internal sealed partial class NodeConnection +{ + private async ValueTask ReceiveFrameAsync(CancellationToken ct) + { + var writer = new ArrayBufferWriter(1024); + while (true) + { + Memory buffer = writer.GetMemory(4096); + ValueWebSocketReceiveResult result = await _socket.ReceiveAsync(buffer, ct); + if (result.MessageType == WebSocketMessageType.Close) + { + return null; + } + + writer.Advance(result.Count); + if (writer.WrittenCount > _host.Options.MaxFrameBytes) + { + _closeReason ??= "frame too large"; + await _socket.CloseOutputAsync(WebSocketCloseStatus.MessageTooBig, "frame too large", CancellationToken.None); + return null; + } + + if (result.EndOfMessage) + { + return new Frame(result.MessageType == WebSocketMessageType.Text, writer.WrittenSpan.ToArray()); + } + } + } + + private static bool TryParse(byte[] data, [System.Diagnostics.CodeAnalysis.NotNullWhen(true)] out RpcMessage? message) + { + try + { + message = JsonSerializer.Deserialize(data, FleetRpcContext.Default.RpcMessage); + return message is not null; + } + catch (JsonException) + { + message = null; + return false; + } + } + + private static bool TryDecode(string text, out byte[] data) + { + try + { + data = Convert.FromBase64String(text); + return true; + } + catch (FormatException) + { + data = []; + return false; + } + } + + private static string StatusName(NodeStatus status) => status switch + { + NodeStatus.Active => "active", + NodeStatus.Pending => "pending", + _ => "disabled", + }; + + private sealed record Frame(bool IsText, byte[] Data); +} + + +[System.Text.Json.Serialization.JsonSerializable(typeof(EmptyObject))] +internal sealed partial class FleetEmptyObjectContext : System.Text.Json.Serialization.JsonSerializerContext; + +internal sealed class EmptyObject; + +internal static class FleetEmptyObject +{ + public static System.Text.Json.Serialization.Metadata.JsonTypeInfo Info => FleetEmptyObjectContext.Default.EmptyObject; +} diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.cs index fca7c46..1d32de7 100644 --- a/backend/fleet/src/Fleet.Server.Link/NodeConnection.cs +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.cs @@ -1,11 +1,7 @@ -using System.Buffers; using System.Collections.Concurrent; using System.Net.WebSockets; -using System.Security.Cryptography; using System.Threading.Channels; -using System.Text.Json; using Fleet.Core; -using Fleet.Identity; using Fleet.Rpc; namespace Fleet.Server; @@ -14,7 +10,7 @@ namespace Fleet.Server; /// 一个节点的 WebSocket 会话:挑战 → 签名认证 → hello → 状态推送 / 收件箱 / 产品 RPC。 /// 收到的消息按顺序处理,所以产品的处理函数不能等待“向同一个节点发起的请求”,否则会死锁。 /// -internal sealed class NodeConnection : IDisposable +internal sealed partial class NodeConnection : IDisposable { private static readonly TimeSpan s_authTimeout = TimeSpan.FromSeconds(15); private static readonly TimeSpan s_presenceRefresh = TimeSpan.FromSeconds(60); @@ -34,6 +30,7 @@ internal sealed class NodeConnection : IDisposable private NodeSessionInfo? _session; private string? _closeReason; private bool _helloDone; + private int _violations; public NodeConnection(FleetLinkHost host, WebSocket socket) { @@ -140,590 +137,4 @@ internal sealed class NodeConnection : IDisposable _lifetime.Dispose(); _sendLock.Dispose(); } - - public void SignalStateChanged() => _stateSignal.Writer.TryWrite(true); - - /// - /// 服务端发起的关闭:发送 Close 帧,等对端确认之后接收循环自然结束。 - /// 注意不能直接取消接收的令牌——取消 ReceiveAsync 会中止整个 WebSocket,对端拿不到关闭原因;取消只作为超时兜底。 - /// - public async ValueTask CloseAsync(string reason) - { - _closeReason ??= reason; - try - { - await _sendLock.WaitAsync(CancellationToken.None); - } - catch (ObjectDisposedException) - { - return; - } - - try - { - if (_socket.State == WebSocketState.Open) - { - await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, reason, CancellationToken.None); - } - } - catch (WebSocketException) - { - // 对端已经断开 - } - finally - { - _sendLock.Release(); - } - - _lifetime.CancelAfter(s_closeTimeout); - } - - public async ValueTask RevokeAsync(string reason, CancellationToken ct) - { - await SendAsync(RpcMessage.Notification(FleetMethods.Revoked, new RevokedParams { Reason = reason }, FleetRpcContext.Default.RevokedParams), ct); - await CloseAsync(reason); - } - - public async ValueTask SendAsync(RpcMessage message, CancellationToken ct) - { - byte[] bytes = JsonSerializer.SerializeToUtf8Bytes(message, FleetRpcContext.Default.RpcMessage); - try - { - await _sendLock.WaitAsync(ct); - } - catch (ObjectDisposedException) - { - return false; - } - - try - { - if (_socket.State != WebSocketState.Open) - { - return false; - } - - await _socket.SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, ct); - return true; - } - catch (WebSocketException) - { - return false; - } - finally - { - _sendLock.Release(); - } - } - - public async ValueTask> RequestAsync(string method, JsonElement parameters, CancellationToken ct) - { - long id = Interlocked.Increment(ref _nextRequestId); - var waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - _pending[id] = waiter; - try - { - var request = new RpcMessage { Id = id, Method = method, Params = parameters }; - if (!await SendAsync(request, ct)) - { - return Result.Fail(FleetErrors.NodeOffline, "node is not connected"); - } - - using var timeout = new CancellationTokenSource(_host.Options.RequestTimeout, _host.Time); - using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); - RpcMessage response = await waiter.Task.WaitAsync(linked.Token); - return response.Error is not null - ? Result.Fail(response.Error.ToFleetError()) - : Result.Ok(response.Result ?? default); - } - catch (OperationCanceledException) when (!ct.IsCancellationRequested) - { - return Result.Fail(FleetErrors.Timeout, $"{method} timed out"); - } - finally - { - _pending.TryRemove(id, out _); - } - } - - // ───────────────────────── 认证 ───────────────────────── - - private async Task AuthenticateAsync(CancellationToken ct) - { - string nonce = Convert.ToBase64String(RandomNumberGenerator.GetBytes(32)); - long expiresAtMs = _host.Time.GetUtcNow().Add(_host.Options.ChallengeTtl).ToUnixTimeMilliseconds(); - await SendAsync( - RpcMessage.Notification(FleetMethods.Challenge, new ChallengeParams { Nonce = nonce, ExpiresAtMs = expiresAtMs }, FleetRpcContext.Default.ChallengeParams), - ct); - - using var timeout = new CancellationTokenSource(s_authTimeout, _host.Time); - using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); - Frame? frame; - try - { - frame = await ReceiveFrameAsync(linked.Token); - } - catch (OperationCanceledException) when (!ct.IsCancellationRequested) - { - await RejectAsync(null, "auth timeout", ct); - return false; - } - - if (frame is not { IsText: true } || !TryParse(frame.Data, out RpcMessage? message) || message.Method != FleetMethods.Auth || message.Id is null) - { - await RejectAsync(null, "expected fleet.auth", ct); - return false; - } - - AuthParams? auth = null; - try - { - auth = message.Params?.Deserialize(FleetRpcContext.Default.AuthParams); - } - catch (JsonException) - { - // 按缺失处理 - } - - NodeRecord? node = auth is null ? null : await _host.Store.GetNodeAsync(new NodeId(auth.NodeId), ct); - bool signatureOk = node is not null - && _host.Time.GetUtcNow().ToUnixTimeMilliseconds() <= expiresAtMs - && TryDecode(auth!.Signature, out byte[] signature) - && NodeIdentity.Verify(node.PublicKey, signature, SignaturePurpose.Auth, auth.NodeId, nonce); - if (!signatureOk) - { - // 节点不存在与签名错误返回同样的错误,不泄露节点是否存在。 - await RejectAsync(message.Id, "authentication failed", ct); - return false; - } - - _node = node; - if (node!.Status == NodeStatus.Disabled) - { - await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.NodeDisabled, "node is disabled")), ct); - await RevokeAsync("disabled", ct); - return false; - } - - _session = new NodeSessionInfo(node.Id, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), node.NodeVersion, [], null); - await SendAsync( - RpcMessage.Success(message.Id.Value, new AuthResult { Status = StatusName(node.Status) }, FleetRpcContext.Default.AuthResult), - ct); - return true; - } - - private async Task RejectAsync(long? id, string reason, CancellationToken ct) - { - FleetLog.AuthRejected(_host.Logger, reason); - _closeReason ??= reason; - if (id is not null) - { - await SendAsync(RpcMessage.Failure(id, new FleetError(FleetErrors.Unauthorized, "authentication failed")), ct); - } - - if (_socket.State == WebSocketState.Open) - { - try - { - await _socket.CloseOutputAsync(WebSocketCloseStatus.PolicyViolation, "unauthorized", ct); - } - catch (WebSocketException) - { - // 对端已经断开 - } - } - } - - // ───────────────────────── 接收循环 ───────────────────────── - - private async Task ReceiveLoopAsync(CancellationToken ct) - { - while (!ct.IsCancellationRequested) - { - using var idle = new CancellationTokenSource(_host.Options.HeartbeatTimeout, _host.Time); - using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, idle.Token); - Frame? frame; - try - { - frame = await ReceiveFrameAsync(linked.Token); - } - catch (OperationCanceledException) when (idle.IsCancellationRequested && !ct.IsCancellationRequested) - { - _closeReason ??= "heartbeat timeout"; - return; - } - - if (frame is null) - { - _closeReason ??= "closed by peer"; - return; - } - - if (frame.IsText) - { - if (TryParse(frame.Data, out RpcMessage? message)) - { - await HandleMessageAsync(message, ct); - } - } - else - { - await HandleBinaryAsync(frame.Data, ct); - } - } - } - - private async Task HandleMessageAsync(RpcMessage message, CancellationToken ct) - { - if (message.IsResponse) - { - if (_pending.TryGetValue(message.Id!.Value, out TaskCompletionSource? waiter)) - { - waiter.TrySetResult(message); - } - - return; - } - - switch (message.Method) - { - case FleetMethods.Hello: - await HandleHelloAsync(message, ct); - return; - case FleetMethods.Heartbeat: - await HandleHeartbeatAsync(message, ct); - return; - } - - if (_node!.Status != NodeStatus.Active) - { - if (message.Id is not null) - { - await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.NodePending, "node is not approved yet")), ct); - } - - return; - } - - var context = new NodeCallContext(_node, _session!, _host); - RpcMessage? response; - try - { - response = await _host.Router.DispatchAsync(context, message, ct); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - throw; - } - - if (response is not null) - { - await SendAsync(response, ct); - } - } - - private async Task HandleHelloAsync(RpcMessage message, CancellationToken ct) - { - HelloParams? hello = null; - try - { - hello = message.Params?.Deserialize(FleetRpcContext.Default.HelloParams); - } - catch (JsonException) - { - // 按缺失处理 - } - - if (hello is null) - { - if (message.Id is not null) - { - await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.InvalidParams, "invalid hello")), ct); - } - - return; - } - - long nowMs = _host.Time.GetUtcNow().ToUnixTimeMilliseconds(); - _session = _session! with - { - NodeVersion = hello.NodeVersion, - Capabilities = hello.Capabilities, - HelloMetaJson = hello.Meta?.GetRawText(), - }; - _node = _node! with { NodeVersion = hello.NodeVersion }; - _helloDone = true; - - await UpdatePresenceAsync(online: true, ct); - await FireAsync(NodeEventKind.Hello, ct, hello); - - if (message.Id is not null) - { - await SendAsync( - RpcMessage.Success( - message.Id.Value, - new HelloResult { ServerTimeMs = nowMs, ClockOffsetMs = hello.NowUtcMs - nowMs }, - FleetRpcContext.Default.HelloResult), - ct); - } - - SignalStateChanged(); - } - - private async Task HandleHeartbeatAsync(RpcMessage message, CancellationToken ct) - { - JsonElement? extra = null; - try - { - extra = message.Params?.Deserialize(FleetRpcContext.Default.HeartbeatParams)?.Extra; - } - catch (JsonException) - { - // 忽略格式不正确的附加字段 - } - - long now = _host.Time.GetTimestamp(); - if (_host.Time.GetElapsedTime(Interlocked.Read(ref _lastPresenceTicks), now) >= s_presenceRefresh) - { - await UpdatePresenceAsync(online: true, ct); - } - - await FireAsync(NodeEventKind.Heartbeat, ct, hello: null, extra); - if (message.Id is not null) - { - await SendAsync(new RpcMessage { Id = message.Id, Result = JsonSerializer.SerializeToElement(new EmptyObject(), FleetEmptyObject.Info) }, ct); - } - } - - private async Task HandleBinaryAsync(byte[] data, CancellationToken ct) - { - if (_node!.Status != NodeStatus.Active - || !OutboxFrame.TryParse(data, out byte channel, out Guid msgId, out ReadOnlyMemory payload) - || !_host.Inboxes.TryGetValue(channel, out IInbox? inbox)) - { - return; - } - - InboxOutcome outcome; - try - { - outcome = await inbox.HandleAsync(_node, msgId, payload, ct); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - throw; - } - catch (Exception ex) - { - // 没有持久化就不确认,节点会重发。 - FleetLog.InboxFailed(_host.Logger, ex, _node.Id.Value, channel); - return; - } - - if (outcome == InboxOutcome.Rejected) - { - return; - } - - var ack = new OutboxAckParams(); - string id = msgId.ToString(); - if (outcome == InboxOutcome.Duplicate) - { - ack.Duplicates.Add(id); - } - else - { - ack.Ids.Add(id); - } - - await SendAsync(RpcMessage.Notification(FleetMethods.OutboxAck, ack, FleetRpcContext.Default.OutboxAckParams), ct); - } - - // ───────────────────────── 状态推送 ───────────────────────── - - private async Task StatePushLoopAsync(CancellationToken ct) - { - while (await _stateSignal.Reader.WaitToReadAsync(ct)) - { - _stateSignal.Reader.TryRead(out _); - bool complete = await PushStateAsync(ct); - if (!complete) - { - await Task.Delay(s_statePushRetry, _host.Time, ct); - SignalStateChanged(); - } - } - } - - private async Task PushStateAsync(CancellationToken ct) - { - NodeRecord? fresh = await _host.Store.GetNodeAsync(_node!.Id, ct); - if (fresh is null) - { - return true; - } - - _node = fresh with { NodeVersion = _node.NodeVersion ?? fresh.NodeVersion }; - if (fresh.Status == NodeStatus.Disabled) - { - await RevokeAsync("disabled", ct); - return true; - } - - if (fresh.Status != NodeStatus.Active || !_helloDone) - { - // pending 不下发状态;还没有 hello 的连接等 hello 之后再推送。 - return true; - } - - bool complete = true; - foreach (IStateProvider provider in _host.StateProviders) - { - StateDocument? document = await provider.GetAsync(_node, ct); - if (document is null || (_lastPushedVersion.TryGetValue(provider.Type, out long last) && last == document.Version)) - { - continue; - } - - Result raw = await RequestAsync( - FleetMethods.StateSync, - JsonSerializer.SerializeToElement( - new StateSyncParams { Type = provider.Type, Version = document.Version, Doc = document.Doc }, - FleetRpcContext.Default.StateSyncParams), - ct); - if (!raw.TryGet(out JsonElement value, out FleetError? error)) - { - FleetLog.StatePushFailed(_host.Logger, _node.Id.Value, provider.Type, error.Code.Value); - complete = false; - continue; - } - - StateSyncResult? result = null; - try - { - result = value.Deserialize(FleetRpcContext.Default.StateSyncResult); - } - catch (JsonException) - { - // 按缺失处理 - } - - if (result is null) - { - complete = false; - continue; - } - - _lastPushedVersion[provider.Type] = document.Version; - await provider.OnAppliedAsync(_node, document.Version, result, ct); - } - - return complete; - } - - // ───────────────────────── 帮助方法 ───────────────────────── - - private async Task UpdatePresenceAsync(bool online, CancellationToken ct) - { - Interlocked.Exchange(ref _lastPresenceTicks, _host.Time.GetTimestamp()); - await _host.Store.UpdatePresenceAsync( - _node!.Id, - new Presence(online, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), _session?.NodeVersion, _session?.Capabilities ?? [], _session?.HelloMetaJson), - ct); - } - - private async Task FireAsync(NodeEventKind kind, CancellationToken ct, HelloParams? hello = null, JsonElement? extra = null) - { - var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra); - foreach (INodeEventSink sink in _host.Sinks) - { - try - { - await sink.OnEventAsync(nodeEvent, ct); - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) - { - throw; - } - catch (Exception ex) - { - // 产品的事件处理失败不能影响连接。 - FleetLog.SinkFailed(_host.Logger, ex, kind.ToString(), _node!.Id.Value); - } - } - } - - private async ValueTask ReceiveFrameAsync(CancellationToken ct) - { - var writer = new ArrayBufferWriter(1024); - while (true) - { - Memory buffer = writer.GetMemory(4096); - ValueWebSocketReceiveResult result = await _socket.ReceiveAsync(buffer, ct); - if (result.MessageType == WebSocketMessageType.Close) - { - return null; - } - - writer.Advance(result.Count); - if (writer.WrittenCount > _host.Options.MaxFrameBytes) - { - _closeReason ??= "frame too large"; - await _socket.CloseOutputAsync(WebSocketCloseStatus.MessageTooBig, "frame too large", CancellationToken.None); - return null; - } - - if (result.EndOfMessage) - { - return new Frame(result.MessageType == WebSocketMessageType.Text, writer.WrittenSpan.ToArray()); - } - } - } - - private static bool TryParse(byte[] data, [System.Diagnostics.CodeAnalysis.NotNullWhen(true)] out RpcMessage? message) - { - try - { - message = JsonSerializer.Deserialize(data, FleetRpcContext.Default.RpcMessage); - return message is not null; - } - catch (JsonException) - { - message = null; - return false; - } - } - - private static bool TryDecode(string text, out byte[] data) - { - try - { - data = Convert.FromBase64String(text); - return true; - } - catch (FormatException) - { - data = []; - return false; - } - } - - private static string StatusName(NodeStatus status) => status switch - { - NodeStatus.Active => "active", - NodeStatus.Pending => "pending", - _ => "disabled", - }; - - private sealed record Frame(bool IsText, byte[] Data); -} - -[System.Text.Json.Serialization.JsonSerializable(typeof(EmptyObject))] -internal sealed partial class FleetEmptyObjectContext : System.Text.Json.Serialization.JsonSerializerContext; - -internal sealed class EmptyObject; - -internal static class FleetEmptyObject -{ - public static System.Text.Json.Serialization.Metadata.JsonTypeInfo Info => FleetEmptyObjectContext.Default.EmptyObject; } diff --git a/backend/fleet/tests/Fleet.Tests/IdentityAndRpcTests.cs b/backend/fleet/tests/Fleet.Tests/IdentityAndRpcTests.cs index d70b033..e5f8476 100644 --- a/backend/fleet/tests/Fleet.Tests/IdentityAndRpcTests.cs +++ b/backend/fleet/tests/Fleet.Tests/IdentityAndRpcTests.cs @@ -67,7 +67,7 @@ public sealed class IdentityAndRpcTests [Fact] public async Task Router_DispatchesTypedRequests_AndReportsStructuredErrors() { - var router = new RpcRouter() + var router = new RpcRouter(RpcSide.Listener, TestTables.Of(TestTables.Req("test.add"), TestTables.Req("test.boom"), TestTables.Req("test.nope"))) .On("test.add", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (caller, p, _) => ValueTask.FromResult(p.A < 0 ? Result.Fail(FleetErrors.Forbidden, "negative") : Result.Ok(new AddResult { Sum = p.A + p.B, Caller = caller }))) .On("test.boom", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => throw new InvalidOperationException("secret detail")); @@ -92,7 +92,7 @@ public sealed class IdentityAndRpcTests public async Task Router_HandlerExceptions_AreContained_AndNeverLeakDetails() { string? logged = null; - var router = new RpcRouter((method, ex) => logged = method + ":" + ex.Message) + var router = new RpcRouter(RpcSide.Listener, TestTables.Of(TestTables.Req("test.boom")), onUnhandled: (method, ex) => logged = method + ":" + ex.Message) .On("test.boom", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => throw new InvalidOperationException("secret detail")); RpcMessage reply = (await router.DispatchAsync("a", Request(1, "test.boom", """{"a":1,"b":1}"""), CancellationToken.None))!; @@ -105,7 +105,7 @@ public sealed class IdentityAndRpcTests public async Task Router_Notifications_ProduceNoResponse_EvenOnFailure() { int calls = 0; - var router = new RpcRouter().OnNotify("test.note", EchoContext.Default.AddParams, (_, _, _) => + var router = new RpcRouter(RpcSide.Listener, TestTables.Of(TestTables.Note("test.note"), TestTables.Note("test.unknown"))).OnNotify("test.note", EchoContext.Default.AddParams, (_, _, _) => { calls++; return ValueTask.CompletedTask; diff --git a/backend/fleet/tests/Fleet.Tests/LinkTests.cs b/backend/fleet/tests/Fleet.Tests/LinkTests.cs index 588f40c..ecbbd43 100644 --- a/backend/fleet/tests/Fleet.Tests/LinkTests.cs +++ b/backend/fleet/tests/Fleet.Tests/LinkTests.cs @@ -227,8 +227,9 @@ public sealed class LinkTests { (TlsFrameTransport left, TlsFrameTransport right) = await PairAsync(); RpcSession? rightSession = null; - var leftRouter = new RpcRouter().On("left.double", LinkJson.Default.Number, LinkJson.Default.Number, (_, p, _) => ValueTask.FromResult(Result.Ok(new Number { Value = p.Value * 2 }))); - var rightRouter = new RpcRouter().On( + MethodTable table = TestTables.Of(TestTables.Req("left.double", RpcSide.Listener), TestTables.Req("right.relay"), TestTables.Req("right.unhandled")); + var leftRouter = new RpcRouter(RpcSide.Dialer, table).On("left.double", LinkJson.Default.Number, LinkJson.Default.Number, (_, p, _) => ValueTask.FromResult(Result.Ok(new Number { Value = p.Value * 2 }))); + var rightRouter = new RpcRouter(RpcSide.Listener, table).On( "right.relay", LinkJson.Default.Number, LinkJson.Default.Number, async (_, p, ct) => { @@ -248,7 +249,7 @@ public sealed class LinkTests Result result = await leftSession.RequestAsync("right.relay", new Number { Value = 21 }, LinkJson.Default.Number, LinkJson.Default.Number, default); Assert.Equal(43, result.Value.Value); - Result unknown = await leftSession.RequestAsync("nope", null, default); + Result unknown = await leftSession.RequestAsync("right.unhandled", null, default); Assert.Equal("fleet_unsupported", unknown.Error.Code.Value); await leftSession.CloseAsync("bye", default); @@ -262,13 +263,13 @@ public sealed class LinkTests { (TlsFrameTransport left, TlsFrameTransport right) = await PairAsync(); await using var silent = right; // 对端不读也不回 - await using var session = new RpcSession(left, new RpcRouter(), "x", TimeProvider.System, new RpcSessionOptions { RequestTimeout = TimeSpan.FromMilliseconds(200) }); + await using var session = new RpcSession(left, new RpcRouter(RpcSide.Dialer, TestTables.Of(TestTables.Req("test.any"))), "x", TimeProvider.System, new RpcSessionOptions { RequestTimeout = TimeSpan.FromMilliseconds(200) }); Task run = session.RunAsync(default); - Result timedOut = await session.RequestAsync("any", null, default); + Result timedOut = await session.RequestAsync("test.any", null, default); Assert.Equal(FleetErrors.Timeout, timedOut.Error.Code); - Task> pending = session.RequestAsync("any", null, default, TimeSpan.FromSeconds(30)).AsTask(); + Task> pending = session.RequestAsync("test.any", null, default, TimeSpan.FromSeconds(30)).AsTask(); await silent.CloseAsync("gone", default); Assert.Equal(FleetErrors.Timeout, (await pending.WaitAsync(TimeSpan.FromSeconds(5))).Error.Code); await run.WaitAsync(TimeSpan.FromSeconds(5)); @@ -372,7 +373,8 @@ public sealed class LinkTests public async Task PeerSession_CarriesRpcBothWays_AfterTheHandshake() { using NodeIdentity probe = NodeIdentity.Generate(); - var router = new RpcRouter().On("agent.echo", LinkJson.Default.Number, LinkJson.Default.Number, (_, p, _) => ValueTask.FromResult(Result.Ok(new Number { Value = p.Value + 1 }))); + MethodTable table = TestTables.Of(TestTables.Req("agent.echo")); + var router = new RpcRouter(RpcSide.Listener, table).On("agent.echo", LinkJson.Default.Number, LinkJson.Default.Number, (_, p, _) => ValueTask.FromResult(Result.Ok(new Number { Value = p.Value + 1 }))); await using var agent = new PeerFixture( _ => probe.PublicKeySpki, async (info, transport, ct) => @@ -383,7 +385,7 @@ public sealed class LinkTests await agent.Listener.Started.Task; (_, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, probe, "prb_1", null, 1 << 20, TimeProvider.System, default); - await using var client = new RpcSession(transport, new RpcRouter(), "probe", TimeProvider.System); + await using var client = new RpcSession(transport, new RpcRouter(RpcSide.Dialer, table), "probe", TimeProvider.System); Task run = client.RunAsync(default); Result result = await client.RequestAsync("agent.echo", new Number { Value = 41 }, LinkJson.Default.Number, LinkJson.Default.Number, default); Assert.Equal(42, result.Value.Value); diff --git a/backend/fleet/tests/Fleet.Tests/MethodDirectionTests.cs b/backend/fleet/tests/Fleet.Tests/MethodDirectionTests.cs new file mode 100644 index 0000000..d87c821 --- /dev/null +++ b/backend/fleet/tests/Fleet.Tests/MethodDirectionTests.cs @@ -0,0 +1,200 @@ +using System.Text.Json; +using System.Threading.Channels; +using Fleet.Core; +using Fleet.Rpc; + +namespace Fleet.Tests; + +/// 协议方向声明:路由器是唯一卡口,注册、分发、发送三处都核对声明表。 +public sealed class MethodDirectionTests +{ + private static readonly MethodTable s_table = TestTables.Of( + MethodSpec.Request("t.ping", RpcSide.Node), + MethodSpec.Notification("t.note", RpcSide.Node, "agent"), + MethodSpec.Request("t.push", RpcSide.Server), + MethodSpec.Request("t.inline", RpcSide.Node).AsInline(), + // 同一个方法名在对等面里另有发起方:按(平面,名字)区分 + MethodSpec.Request("t.ping", RpcSide.Listener)); + + private static readonly JsonElement s_empty = JsonDocument.Parse("{}").RootElement.Clone(); + + private static RpcMessage Request(string method) => new() { Id = 1, Method = method, Params = s_empty }; + + private static RpcMessage Notification(string method) => new() { Method = method, Params = s_empty }; + + private static RpcRouter Server(string? localRole = null) => new(RpcSide.Server, s_table, localRole, callerRole: role => role); + + [Fact] + public void Registration_RefusesUndeclaredMethods_WrongDirection_WrongKind_AndWrongRole() + { + RpcRouter server = Server(); + Assert.Throws(() => server.On("t.unknown", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => throw new NotSupportedException())); + // t.push 是服务端发给节点的:服务端这一侧不能处理它 + Assert.Throws(() => server.On("t.push", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => throw new NotSupportedException())); + // t.ping 声明成请求:不能按通知注册 + Assert.Throws(() => server.OnNotify("t.ping", EchoContext.Default.AddParams, (_, _, _) => ValueTask.CompletedTask)); + + // 节点这一侧:t.note 只给 agent 角色,probe 注册不了;t.push 是它该处理的 + var probe = new RpcRouter(RpcSide.Node, s_table, "probe"); + probe.On("t.push", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => throw new NotSupportedException()); + Assert.Throws(() => new RpcRouter(RpcSide.Server, s_table, "probe").OnNotify("t.note", EchoContext.Default.AddParams, (_, _, _) => ValueTask.CompletedTask)); + + server.On("t.ping", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => ValueTask.FromResult(Result.Ok(new AddResult()))); + } + + [Fact] + public async Task Dispatch_RefusesWhatThePeerMayNotInitiate_AndReportsAViolation() + { + int calls = 0; + RpcRouter server = Server() + .On("t.ping", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => + { + calls++; + return ValueTask.FromResult(Result.Ok(new AddResult())); + }) + .OnNotify("t.note", EchoContext.Default.AddParams, (_, _, _) => + { + calls++; + return ValueTask.CompletedTask; + }); + + // 正常 + RpcDispatch ok = await server.DispatchCheckedAsync("agent", Request("t.ping"), default); + Assert.Null(ok.Violation); + Assert.NotNull(ok.Response!.Result); + + // 没有声明的方法 + RpcDispatch undeclared = await server.DispatchCheckedAsync("agent", Request("t.nope"), default); + Assert.NotNull(undeclared.Violation); + Assert.Equal("fleet_protocol_violation", undeclared.Response!.Error!.Data!.Code); + + // 节点冒充服务端:t.push 只能由服务端发起 + RpcDispatch backwards = await server.DispatchCheckedAsync("agent", Request("t.push"), default); + Assert.NotNull(backwards.Violation); + Assert.Equal("fleet_protocol_violation", backwards.Response!.Error!.Data!.Code); + + // 请求和通知用反了 + Assert.NotNull((await server.DispatchCheckedAsync("agent", Notification("t.ping"), default)).Violation); + RpcDispatch asRequest = await server.DispatchCheckedAsync("agent", Request("t.note"), default); + Assert.NotNull(asRequest.Violation); + Assert.NotNull(asRequest.Response); + + // 违规的消息到不了处理函数;通知违规不回响应 + Assert.Equal(1, calls); + RpcDispatch silentViolation = await server.DispatchCheckedAsync("agent", Notification("t.nope"), default); + Assert.NotNull(silentViolation.Violation); + Assert.Null(silentViolation.Response); + } + + [Fact] + public async Task Dispatch_ADeclaredMethodWithoutAHandler_IsUnsupported_NotAViolation() + { + RpcDispatch outcome = await Server().DispatchCheckedAsync("agent", Request("t.ping"), default); + Assert.Null(outcome.Violation); + Assert.Equal("fleet_unsupported", outcome.Response!.Error!.Data!.Code); + } + + [Fact] + public async Task Dispatch_ChecksTheCallersRole() + { + RpcRouter server = Server().OnNotify("t.note", EchoContext.Default.AddParams, (_, _, _) => ValueTask.CompletedTask); + Assert.Null((await server.DispatchCheckedAsync("agent", Notification("t.note"), default)).Violation); + Assert.NotNull((await server.DispatchCheckedAsync("probe", Notification("t.note"), default)).Violation); + Assert.NotNull((await server.DispatchCheckedAsync(null!, Notification("t.note"), default)).Violation); + } + + [Fact] + public void CheckSend_RefusesWhatThisSideMayNotInitiate() + { + RpcRouter server = Server(); + server.CheckSend("t.push", MethodKind.Request); + Assert.Throws(() => server.CheckSend("t.ping", MethodKind.Request)); // 那是节点发起的 + Assert.Throws(() => server.CheckSend("t.push", MethodKind.Notification)); // 类型不对 + Assert.Throws(() => server.CheckSend("t.unknown", MethodKind.Request)); + + // 对端的角色也要允许 + var node = new RpcRouter(RpcSide.Node, s_table, "agent"); + node.CheckSend("t.note", MethodKind.Notification, "agent"); + Assert.Throws(() => node.CheckSend("t.note", MethodKind.Notification, "probe")); + } + + [Fact] + public void MethodTable_RefusesDuplicates_AndReportsMissingHandlers() + { + Assert.Throws(() => TestTables.Of(TestTables.Req("a"), TestTables.Note("a"))); + Assert.Throws(() => new MethodTable([], [new ChannelSpec(16, RpcSide.Node), new ChannelSpec(16, RpcSide.Server)])); + + // 服务端要处理 t.ping 和 t.note;t.inline 在握手阶段直接处理,不算缺 + Assert.Equal(["t.note", "t.ping"], s_table.MissingHandlers(RpcSide.Server, null, [])); + Assert.Equal(["t.ping"], s_table.MissingHandlers(RpcSide.Server, "probe", [])); + Assert.Equal(["t.note"], s_table.MissingHandlers(RpcSide.Server, "agent", ["t.ping"])); + Assert.Empty(s_table.MissingHandlers(RpcSide.Server, "agent", ["t.ping", "t.note"])); + // 同名方法在对等面的接收方是拨号方?不:t.ping 在对等面由监听方发起,拨号方处理 + Assert.Equal(["t.ping"], s_table.MissingHandlers(RpcSide.Dialer, null, [])); + } + + [Fact] + public async Task Session_ClosesTheConnection_AfterTooManyViolations() + { + (ChannelTransport a, ChannelTransport b) = ChannelTransport.Pair(); + RpcRouter server = Server().On("t.ping", EchoContext.Default.AddParams, EchoContext.Default.AddResult, (_, _, _) => ValueTask.FromResult(Result.Ok(new AddResult()))); + await using var session = new RpcSession(b, server, "agent", TimeProvider.System, new RpcSessionOptions { MaxViolations = 3 }); + Task run = session.RunAsync(default); + + for (int i = 1; i <= 6; i++) + { + await a.SendAsync(new RpcFrame(FrameKind.Text, JsonSerializer.SerializeToUtf8Bytes(new RpcMessage { Id = i, Method = "t.nope", Params = s_empty }, FleetRpcContext.Default.RpcMessage)), default); + } + + // 违规的请求先收到失败响应,超过上限之后连接被关闭 + int failures = 0; + while (await a.ReceiveAsync(default).AsTask().WaitAsync(TimeSpan.FromSeconds(5)) is { } frame) + { + if (JsonSerializer.Deserialize(frame.Data.Span, FleetRpcContext.Default.RpcMessage)?.Error?.Data?.Code == "fleet_protocol_violation") + { + failures++; + } + } + + Assert.InRange(failures, 3, 6); + await a.CloseAsync("done", default); // 对端确认关闭之后,会话的接收循环正常结束 + await run.WaitAsync(TimeSpan.FromSeconds(5)); + } + + /// 进程内的一对传输:一端发的帧从另一端收到;关闭时对端收到 null。 + private sealed class ChannelTransport : IRpcTransport + { + private readonly Channel _in; + private readonly Channel _out; + + private ChannelTransport(Channel input, Channel output) + { + _in = input; + _out = output; + } + + public static (ChannelTransport A, ChannelTransport B) Pair() + { + Channel ab = Channel.CreateUnbounded(); + Channel ba = Channel.CreateUnbounded(); + return (new ChannelTransport(ba, ab), new ChannelTransport(ab, ba)); + } + + public ValueTask SendAsync(RpcFrame frame, CancellationToken ct) + { + _out.Writer.TryWrite(frame); + return ValueTask.CompletedTask; + } + + public async ValueTask ReceiveAsync(CancellationToken ct) => + await _in.Reader.WaitToReadAsync(ct) && _in.Reader.TryRead(out RpcFrame? frame) ? frame : null; + + public ValueTask CloseAsync(string reason, CancellationToken ct) + { + _out.Writer.TryWrite(null); + return ValueTask.CompletedTask; + } + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +} diff --git a/backend/fleet/tests/Fleet.Tests/RestartScheduleTests.cs b/backend/fleet/tests/Fleet.Tests/RestartScheduleTests.cs index 84c8855..2fbe602 100644 --- a/backend/fleet/tests/Fleet.Tests/RestartScheduleTests.cs +++ b/backend/fleet/tests/Fleet.Tests/RestartScheduleTests.cs @@ -131,7 +131,13 @@ public sealed class RestartScheduleTests Assert.DoesNotContain(log.Lines, l => l.Contains("scheduled_restart", StringComparison.Ordinal)); // 清理:把进程停掉 Assert.True(controller.SetRemote("04:30+08:00")); - time.Advance(); + // 新的计时器是异步注册的:反复推进,直到进程退出 + for (int i = 0; i < 100 && !shell.IsCompleted; i++) + { + time.Advance(); + await Task.Delay(50); + } + await shell.WaitAsync(TimeSpan.FromSeconds(10)); } diff --git a/backend/fleet/tests/Fleet.Tests/TestTables.cs b/backend/fleet/tests/Fleet.Tests/TestTables.cs new file mode 100644 index 0000000..0f51840 --- /dev/null +++ b/backend/fleet/tests/Fleet.Tests/TestTables.cs @@ -0,0 +1,13 @@ +using Fleet.Rpc; + +namespace Fleet.Tests; + +/// 测试里用的方法声明:除非另有说明,方法都由拨号方发给监听方。 +internal static class TestTables +{ + public static MethodTable Of(params MethodSpec[] specs) => new(specs); + + public static MethodSpec Req(string name, RpcSide from = RpcSide.Dialer) => MethodSpec.Request(name, from); + + public static MethodSpec Note(string name, RpcSide from = RpcSide.Dialer) => MethodSpec.Notification(name, from); +} diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index b574a36..90f8ff4 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs @@ -25,6 +25,9 @@ public sealed class AgentConnection private static readonly TimeSpan s_clockInterval = TimeSpan.FromSeconds(30); private static readonly RpcRouter s_router = BuildRouter(); + /// 对等面入站方法的路由器(只读,测试用)。 + public static RpcRouter Router => s_router; + private readonly ConnectionEnv _env; private volatile ConnectionSession? _session; @@ -191,7 +194,7 @@ public sealed class AgentConnection } private static RpcRouter BuildRouter() => - new RpcRouter() + new RpcRouter(RpcSide.Dialer, MoShiroMethodTable.Table) .On( FleetMethods.StateSync, FleetRpcContext.Default.StateSyncParams, diff --git a/backend/probe/src/MoShiro.Probe/ProbeApp.cs b/backend/probe/src/MoShiro.Probe/ProbeApp.cs index f3eba1a..1ac464e 100644 --- a/backend/probe/src/MoShiro.Probe/ProbeApp.cs +++ b/backend/probe/src/MoShiro.Probe/ProbeApp.cs @@ -50,7 +50,7 @@ public sealed class ProbeApp : IDisposable directory.Apply(doc); return ValueTask.FromResult(new StateSyncResult { AppliedVersion = version }); }); - var router = new RpcRouter() + var router = new RpcRouter(RpcSide.Node, MoShiroMethodTable.Table, NodeRoles.Probe) .OnNotify(MoShiroMethods.UpdateOffer, MoShiroJsonContext.Default.UpdateOfferParams, (_, offer, _) => { update.WriteOffer(offer); diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs new file mode 100644 index 0000000..ada3f4a --- /dev/null +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs @@ -0,0 +1,55 @@ +using Fleet.Rpc; + +namespace MoShiro.Protocol; + +/// 节点的角色。服务端按节点 ID 的前缀判断;声明表用它限定哪些角色可以参与某个方法或通道。 +public static class NodeRoles +{ + public const string Agent = "agent"; + public const string Probe = "probe"; + + public static string? Of(string nodeId) => + nodeId.StartsWith(AgentIds.Prefix, StringComparison.Ordinal) ? Agent + : nodeId.StartsWith(ProbeId.Prefix, StringComparison.Ordinal) ? Probe + : null; +} + +/// +/// 所有方法和二进制通道的方向声明,单一真源(框架的 fleet.* 见 )。 +/// 新增一个方法:在 里加名字和参数,在这里加一行声明,再在对应一侧注册处理函数; +/// 声明了但没人处理、或者处理了但没声明,都会被测试和启动时的检查发现。 +/// 同一个方法名只有一个发起方;节点面是服务端和节点(B 和 A、P),对等面是 Probe(拨号方)和 Agent(监听方)。 +/// +public static class MoShiroMethodTable +{ + public static readonly MethodTable Table = MethodTable.Combine( + FleetMethodTable.Table, + new MethodTable( + [ + // 节点面:Backend → 节点 + MethodSpec.Request(MoShiroMethods.TestRun, RpcSide.Server, NodeRoles.Agent), + MethodSpec.Request(MoShiroMethods.TestCancel, RpcSide.Server, NodeRoles.Agent), + MethodSpec.Request(MoShiroMethods.ProbeDisconnect, RpcSide.Server, NodeRoles.Agent), + MethodSpec.Notification(MoShiroMethods.UpdateOffer, RpcSide.Server), + MethodSpec.Notification(MoShiroMethods.UpdateApply, RpcSide.Server), + MethodSpec.Notification(MoShiroMethods.Schedule, RpcSide.Server), + + // 节点面:Agent → Backend + MethodSpec.Notification(MoShiroMethods.TestStatus, RpcSide.Node, NodeRoles.Agent), + MethodSpec.Notification(MoShiroMethods.Event, RpcSide.Node, NodeRoles.Agent), + MethodSpec.Notification(MoShiroMethods.Host, RpcSide.Node, NodeRoles.Agent), + + // 对等面:Probe(拨号方)→ Agent(监听方) + MethodSpec.Request(MoShiroMethods.Clock, RpcSide.Dialer), + + // 对等面:Agent → Probe + MethodSpec.Request(MoShiroMethods.TestPhase, RpcSide.Listener), + MethodSpec.Request(MoShiroMethods.TestPhaseEnd, RpcSide.Listener), + MethodSpec.Notification(MoShiroMethods.TestAbort, RpcSide.Listener), + ], + [ + // 原始块:只有 Agent 往 Backend 发;列块:Probe 经对等会话发给 Agent + new ChannelSpec(MoShiroChannels.RawChunk, RpcSide.Node, [NodeRoles.Agent]), + new ChannelSpec(MoShiroChannels.ColumnBlock, RpcSide.Dialer), + ])); +} diff --git a/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs b/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs index 8cdb3ac..18cc74f 100644 --- a/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs +++ b/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs @@ -22,6 +22,7 @@ public static class GatewayServiceCollectionExtensions .AddState() .AddState() .AddInbox() + .WithMethods(MoShiroMethodTable.Table, node => NodeRoles.Of(node.Id.Value)) .MapMethods(MapMoShiroMethods); return services; } diff --git a/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs b/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs index 465b79a..5fdd240 100644 --- a/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs @@ -168,15 +168,45 @@ public sealed class OperationsAndSecurityTests await agent.SendBinaryAsync([200, 1, 2, 3]); // 未知的二进制通道,长度不够一个完整的头部 await agent.SendBinaryAsync(OutboxFrame.Build(99, Guid.NewGuid(), [1])); // 没有注册的通道 - // 节点不能调用任何“执行”类的方法:Fleet 根本没有这样的路由 + // 节点不能调用任何“执行”类的方法:这些方法没有声明,是协议违规,直接拒绝 foreach (string method in new[] { "fleet.exec", "fleet.shell", "moshiro.exec", "moshiro.write_file", "fleet.update" }) { RpcMessage reply = await agent.RequestRawAsync(method); - Assert.Equal("fleet_unsupported", reply.Error!.Data!.Code); + Assert.Equal("fleet_protocol_violation", reply.Error!.Data!.Code); } - // 连接仍然可用:路由器照常回复 - Assert.Equal("fleet_unsupported", (await agent.RequestRawAsync("fleet.exec")).Error!.Data!.Code); + // 连接仍然可用:违规次数还没到上限,路由器照常回复 + Assert.Equal("fleet_protocol_violation", (await agent.RequestRawAsync("fleet.exec")).Error!.Data!.Code); + } + + [Fact] + public async Task ARepeatOffender_IsDisconnected_AndANodeCannotInitiateWhatOnlyTheBackendMay() + { + await using AgentSimulator agent = await _backend.ConnectAgentAsync(); + await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount); + + // 节点冒充 Backend 发起只有 Backend 才能发的方法:违规,不是「不支持」 + Assert.Equal("fleet_protocol_violation", (await agent.RequestRawAsync(MoShiroMethods.TestRun)).Error!.Data!.Code); + Assert.Equal("fleet_protocol_violation", (await agent.RequestRawAsync("fleet.state.sync")).Error!.Data!.Code); + + // 一个 Probe 都不是的 Agent 发 Probe 才有的通道:节点面的列块通道(17)没有声明,不会进收件箱 + await agent.SendBinaryAsync(OutboxFrame.Build(MoShiroChannels.ColumnBlock, Guid.NewGuid(), [1])); + + // 反复违规:超过上限(默认 20 次)之后连接被服务端关闭 + for (int i = 0; i < 40 && agent.CloseReason is null; i++) + { + try + { + await agent.RequestRawAsync("fleet.exec"); + } + catch (Exception ex) when (ex is TimeoutException or OperationCanceledException or System.Net.WebSockets.WebSocketException or InvalidOperationException) + { + break; + } + } + + await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.CloseReason is not null), because: "a node that keeps violating the protocol is disconnected"); + Assert.Equal("protocol violations", agent.CloseReason); } [Fact] diff --git a/backend/tests/MoShiro.Backend.Tests/ProtocolDeclarationTests.cs b/backend/tests/MoShiro.Backend.Tests/ProtocolDeclarationTests.cs new file mode 100644 index 0000000..5dc920e --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/ProtocolDeclarationTests.cs @@ -0,0 +1,94 @@ +using System.Reflection; +using Fleet.Rpc; +using Fleet.Server; +using Microsoft.Extensions.DependencyInjection; +using MoShiro.Probe; +using MoShiro.Protocol; + +namespace MoShiro.Backend.Tests; + +/// +/// 协议方向声明()和实际代码的对账:每个方法常量都声明了,每一侧该处理的方法都注册了处理函数, +/// 二进制通道都有对应的收件箱,文档里写了每一个方法。新增方法绕不过声明,声明了没人处理也会被发现。 +/// 注册处理函数时路由器本身还会核对方向、类型和角色,不符就在启动时失败。 +/// +[Collection("probe-e2e")] +public sealed class ProtocolDeclarationTests +{ + private static readonly MethodTable s_table = MoShiroMethodTable.Table; + + private readonly BackendFixture _backend; + + public ProtocolDeclarationTests(BackendFixture backend) + { + _backend = backend; + } + + private static IEnumerable ConstantsOf(Type type) => + type.GetFields(BindingFlags.Public | BindingFlags.Static).Where(f => f.IsLiteral && f.FieldType == typeof(string)).Select(f => (string)f.GetRawConstantValue()!); + + [Fact] + public void EveryMethodConstant_IsDeclared() + { + HashSet declared = [.. s_table.Methods.Select(m => m.Name)]; + string[] undeclared = [.. ConstantsOf(typeof(FleetMethods)).Concat(ConstantsOf(typeof(PeerMethods))).Concat(ConstantsOf(typeof(MoShiroMethods))).Where(n => !declared.Contains(n))]; + Assert.Empty(undeclared); + } + + [Fact] + public void EveryDeclaredMethod_IsDocumented() + { + string root = FindRoot(); + string docs = File.ReadAllText(Path.Combine(root, "docs", "protocol.md")) + File.ReadAllText(Path.Combine(root, "backend", "fleet", "README.md")); + string[] missing = [.. s_table.Methods.Select(m => m.Name).Distinct().Where(n => !docs.Contains($"`{n}`", StringComparison.Ordinal))]; + Assert.Empty(missing); + } + + [Fact] + public void Backend_HandlesEveryMethodNodesMayInitiate_AndHasAnInboxForEveryDeclaredChannel() + { + FleetRouter router = _backend.Services.GetRequiredService(); + Assert.Empty(router.Table.MissingHandlers(RpcSide.Server, null, router.Methods)); + + byte[] inboxes = [.. _backend.Services.GetServices().Select(i => i.Channel).Order()]; + byte[] declared = [.. s_table.Channels.Where(c => c.Plane == RpcPlane.Node).Select(c => c.Channel).Order()]; + Assert.Equal(declared, inboxes); + } + + [Fact] + public async Task Agent_HandlesEveryMethodItShould_OnBothPlanes() + { + await using AgentHarness agent = await AgentHarness.StartAsync(_backend); + Assert.Empty(s_table.MissingHandlers(RpcSide.Node, NodeRoles.Agent, agent.App.Link.Router.Methods)); + // 对等面:Agent 是监听方,要处理 Probe 发来的方法 + Assert.Empty(s_table.MissingHandlers(RpcSide.Listener, null, agent.App.PeerRouter.Methods)); + } + + [Fact] + public async Task Probe_HandlesEveryMethodItShould_OnBothPlanes() + { + await using ProbeHarness probe = await ProbeHarness.StartAsync(_backend); + Assert.Empty(s_table.MissingHandlers(RpcSide.Node, NodeRoles.Probe, probe.App.Link.Router.Methods)); + // 对等面:Probe 是拨号方,要处理 Agent 发来的方法 + Assert.Empty(s_table.MissingHandlers(RpcSide.Dialer, null, AgentConnection.Router.Methods)); + } + + [Fact] + public void RolesOfNodeIds_FollowThePrefix() + { + Assert.Equal(NodeRoles.Agent, NodeRoles.Of(AgentIds.Prefix + "_01ABC")); + Assert.Equal(NodeRoles.Probe, NodeRoles.Of(ProbeId.Prefix + "_01ABC")); + Assert.Null(NodeRoles.Of("usr_01ABC")); + } + + private static string FindRoot() + { + string? dir = AppContext.BaseDirectory; + while (dir is not null && !File.Exists(Path.Combine(dir, "docs", "protocol.md"))) + { + dir = Path.GetDirectoryName(dir); + } + + return dir ?? throw new InvalidOperationException("repository root not found"); + } +} diff --git a/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs b/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs index 972bc58..5d0aa39 100644 --- a/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs @@ -313,7 +313,7 @@ public sealed class ProbeSimulator : IAsyncDisposable } private RpcRouter BuildRouter() => - new RpcRouter() + new RpcRouter(RpcSide.Dialer, MoShiroMethodTable.Table) .On( FleetMethods.StateSync, FleetRpcContext.Default.StateSyncParams, diff --git a/docs/architecture.md b/docs/architecture.md index 9cc7ad5..f1f90df 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -88,6 +88,8 @@ CI ──签名的发布清单(镜像 digest)──▶ Backend ──原样 ## 5. 安全边界:纯采集 +> 这一节是现状。「Backend 被攻破、节点被入侵」之后的更严格设计,包括现有的几处缺口(例如 Probe 会拨 Backend 给的任意地址、Agent 的 TCP 端口会对陌生人完成 TLS 握手),见 [security.md](security.md),尚未实现。 + ```text Agent / Probe 能做的 执行编译在二进制中的 Kind;监听和访问本地配置允许的端口与网段;受本地硬上限约束 不能做的 exec / shell / 终端 / 文件读写(配置目录除外)/ 自更新 / 加载外部代码 / 采集主机标识(型号、主机名、IP、进程) @@ -151,6 +153,7 @@ Kind 目录编译在二进制中,节点不接受远程下发代码(安全边 | D20 | 时序响应采用列式格式(`t[]` + `values{}`,时间轴使用 Unix 毫秒) | 由用户确认;体积小,可以直接交给图表库 | | D21 | 国家和运营商只由 Backend 向 IP 归属服务查询(缺省 `https://ip.007995.xyz`,可关闭),只查公网地址;节点不访问任何外部服务;公开接口不返回 IP、城市、ASN | 客户的防火墙常常要审计节点的出站连接;访客拿不到敏感信息(有测试保证) | | D22 | 构建信息(版本、tag、提交、构建时间)由节点自报,访客是否可见由设置 `site.show_build` 控制(默认关);提交号只在管理端 | 展示页可以选择不暴露版本;管理员能看出哪个节点没升级 | +| D23 | 零信任方向:信任锚点从 Backend 移到操作员密钥(名册);Probe 自己生成密钥;Agent 的 TCP 端口预握手敲门;敏感能力放进独立 extension,用操作员签名的能力令牌授权。不做进程沙箱(节点是 1 核 1 GB 的小机器),代码风险靠 PR 评审 | 由用户确认;设计稿见 [security.md](security.md),尚未实现 | ## 9. 仍待确认 diff --git a/docs/protocol.md b/docs/protocol.md index c982a95..7f5dd48 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -22,6 +22,8 @@ MoShiro 的两条控制链路都建立在 Fleet 之上: ## 2. `moshiro.*` 方法 +每个方法的发起方、类型(请求或通知)和允许的角色声明在代码里的 `MoShiroMethodTable`(框架的 `fleet.*` 在 `FleetMethodTable`),路由器据此拒绝没有声明、方向不对或角色不符的消息,回 `fleet_protocol_violation`;下面各表的「方向」列由测试和这张声明表对账。设计见 [security.md](security.md) 第 8 节。 + ### Agent ↔ Backend | 方法 | 方向 | 类型 | 参数与结果 | @@ -31,6 +33,7 @@ MoShiro 的两条控制链路都建立在 Fleet 之上: | `moshiro.test.cancel` | B→A | 请求 | `test_id` | | `moshiro.test.status` | A→B | 通知 | `test_id`、`status`(`running`、`phase_done`、`succeeded`、`failed`、`aborted`)、`phase?`、`error_code?` | | `moshiro.event` | A→B | 通知 | `kind`(`probe_connected`(`detail` 是 `probe_id`,`probe` 里带版本、远端地址和连接时间)、`probe_disconnected`、`endpoint_changed`、`session_rejected`、`clock_step`、`gap`、`cpu_saturated`(节点进程 CPU 饱和,可能影响测量,由 Probe 或 Agent 上报))、`at_utc`、`detail` | +| `moshiro.host` | A→B | 通知 | `HostSnapshot`:主机资源占用的汇总数字(CPU、内存、根分区、网卡累计流量与网速、运行时长;没有型号、主机名、IP),每 `host_interval_s` 秒一次;`host_stats` 为 false 时不发,见 [enrollment.md](enrollment.md) | 心跳的附加字段里有 `probes[]`:Agent 当前连着的每个 Probe(`id`、`version`、`remote`、`since_utc_ms`),每次都是完整快照,Backend 靠它纠正错过的连接事件。点对点握手的认证消息里可以带 Probe 的 `version`(只用于展示)。块头里的 `probe_id` 以 Agent 认证过的会话为准,Probe 自报的值会被覆盖。 @@ -45,7 +48,7 @@ MoShiro 的两条控制链路都建立在 Fleet 之上: | `moshiro.test.abort` | A→P | 通知 | `test_id` | | `fleet.outbox.ack` | 双向 | 通知 | Probe 的列块(通道 17)是至少一次投递,Agent 每收到一个就确认,重复到达的只处理一次 | -会话密钥不需要单独的方法:点对点握手(TLS 固定证书 → `peer.challenge` → Probe 用身份私钥签 `(peer_id, nonce, client_pub, server_pub)` → 两端各自 ECDH + HKDF)直接得到 `session_id` 和 `session_key`,`peer.auth` 的应答只回显 `session_id` 用来互相确认。 +会话密钥不需要单独的方法:点对点握手(TLS 固定证书 → `fleet.peer.challenge` → Probe 用身份私钥签 `(peer_id, nonce, client_pub, server_pub)` → 两端各自 ECDH + HKDF)直接得到 `session_id` 和 `session_key`,`peer.auth` 的应答只回显 `session_id` 用来互相确认。 ### 列块(Probe → Agent,通道 17) diff --git a/docs/security.md b/docs/security.md new file mode 100644 index 0000000..6593da7 --- /dev/null +++ b/docs/security.md @@ -0,0 +1,327 @@ +# 零信任设计 + +> **实现状态:阶段零(协议方向声明表和卡口,第 8 节)已实现,其余是设计稿,尚未实现。** 第 3 节的缺口是对当前代码的核对结果(G7 已解决);第 4 节起是目标设计和实施顺序。第 14 节列出还没定的事。 + +Agent、Backend、Probe 之间互相不信任。这份文档只回答三个问题:现在谁听谁的、被攻破之后最坏会怎样、怎么用密码学把信任从 Backend 身上拿走。纯采集的边界、本地硬上限和签名更新见 [architecture.md](architecture.md) 第 5 节,这里不重复。 + +## 1. 范围 + +**要防的** + +- Backend(或 Backend 前面的 Cloudflare Tunnel、frp、Caddy 等中间环节)被攻破或被劫持之后,给节点下发恶意内容。 +- Agent 或 Probe 自己被入侵之后,去影响 Backend 或对端。 + +**有意不做的** + +- **进程沙箱**(seccomp、pids 和内存上限、出站白名单)。Agent 和 Probe 的定位是 1 核 1 GB 的小机器,不能吃资源。 +- **供应链的多签、冷却期和可复现构建**。代码风险靠 PR 评审降一个等级;发布仍然是单把发布密钥签名(见 [deployment.md](deployment.md))。以后需要时再加,不属于这份设计。 +- 顺带记一条:Agent 的单元现在挂了 `/:/host/root:ro`、`/proc`、`/sys`,被攻破的 Agent 能遍历宿主机。它们多半是不必要的(宿主网络模式下容器自己的 `/proc` 就是宿主的,磁盘用量可以对已挂载的 `/etc/os-release` 做 `statvfs`),属于沙箱范畴,先不动,实现时再核实。 + +## 2. 威胁模型 + +| 编号 | 假设被攻破的是 | 现实来源 | +| --- | --- | --- | +| T1 | Backend,或它前面的中间环节 | 链路经过 Cloudflare Tunnel、frp、Caddy,它们都能看到 wss 的明文 | +| T2 | 某个 Agent(在 VPS 上,旁边还有别的服务) | VPS 被入侵 | +| T3 | 某个 Probe(在客户内网里) | 客户机器被入侵 | +| T4 | 管理员的会话或令牌 | 令牌泄露 | +| T5 | 管理员的浏览器 | XSS,或者节点上报的字符串被渲染 | + +## 3. 现状:节点听谁的,缺口在哪 + +```text + CI 发布私钥(单把) + │ 签名清单 + 管理员 ──令牌/会话──▶ ┌──────────┐ ◀── 原始块、事件、主机数字(没有按节点的配额) + │ Backend │ + └──┬───┬───┘ + 任务集、对端公钥、 │ │ Agent 名单(host:port,Probe 全部照拨) + 重启计划、更新清单 ▼ ▼ + ┌──────┐ ┌──────┐ + │Agent │◀─▶│Probe │ Probe 私钥由 Backend 生成 + └──────┘ └──────┘ +``` + +| # | 缺口 | 依据 | 后果 | +| --- | --- | --- | --- | +| G1 | 节点不验证 Backend 的身份,只靠 TLS 加公共 CA | `Fleet.Link` 里没有证书固定,也没有应用层验证 | 链路上任一环节被攻破,就能给节点下发任务集、Agent 名单、对端公钥、重启计划(更新清单有签名,不受影响) | +| G2 | Probe 的私钥由 Backend 生成 | `ProbeService` 里的 `NodeIdentity.Generate`(创建和轮换) | Backend 接触过 Probe 的私钥,可以冒充它;Agent 的密钥是自己生成的,没有这个问题 | +| G3 | Probe 会拨 Backend 给的任意 `host:port` | `AgentDirectory`、`AgentConnection` 不做任何地址检查;本地 `limits` 只有速率和 pps | 被攻破的 Backend 能让客户内网里的 Probe 去探测内网。固定 SPKI 只能防止建立会话,防不了探测本身 | +| G4 | Agent 的 TCP 端口对任何人完成 TLS 握手并出示证书 | `TcpAcceptor`、`PeerListener` 在 `AcceptTcpClient` 之后直接 `AuthenticateAsServer` | 自签证书、TLS 参数、JARM 都能被扫描器指纹识别,一眼看出这是个服务 | +| G5 | Backend 入库只限制了单块的大小 | `MaxDecodedChunkBytes` 等;没有找到按节点的速率或配额 | 恶意节点可以灌满磁盘 | +| G6 | 对端公钥、Agent 地址全由 Backend 决定 | `TaskSet.Peers`、`ProbeAgentList` | 被攻破的 Backend 能把恶意 Probe 加进 Agent 的白名单,或把 Probe 指向恶意 Agent | +| G7(已解决) | 方法的方向和角色没有统一声明:原来靠「谁注册了处理器」隐式决定 | `NodeConnection.HandleMessageAsync` 只按方法名分发到路由器;文档里的 `B→A`、`A→B` 只存在于 [protocol.md](protocol.md),没有对账 | 新增一个方法时,方向写反、角色写错(例如 Probe 调了只给 Agent 的方法)不会被任何东西发现 | +| G8 | **测量数据没有产生者签名** | `Raw/` 的编解码里没有任何签名或 MAC,块只有 CRC32C;Probe 的列块(通道 17)经 Agent 合并进原始块再上传 | 隧道等中间环节可以改写测量数据;被攻破的 Agent 可以伪造 Probe 侧的时间戳(`t1`、`t4`) | +| G9 | **入库只核对了 `agent_id`** | `IngestService` 在 `header.AgentId != agent` 之后直接采用块头里的 `series`(形如 `line:`),全文没有出现 Probe | 被攻破的 Agent 可以为从没连过它的 Probe 灌入假线路数据 | +| G10 | **没有合理性检查** | `EchoOps` 里没有对回显时间戳做任何一致性检查 | 恶意 Agent 可以谎报 `t2`、`t3`,把自己的处理时延说成任意值,让线路 RTT 好看或难看 | +| G11 | **Probe 的一把密钥多处使用** | [architecture.md](architecture.md) 第 5 节写明同一把用来连 Backend 和连 Agent;已有用途前缀(`auth`、`peer`)做域隔离 | 一处泄露,两处都沦陷 | + +没有问题的部分:UDP 端口对未认证的包静默丢弃,不回应(`DropInvalid`,有测试);Probe 本身没有 TCP 监听;任务有本地硬上限;更新清单有签名和 `seq` 防回滚。 + +## 4. 核心思路:信任锚点从 Backend 移走 + +要防 Backend 被攻破,**下行内容必须由 Backend 之外的密钥授权**。Backend 只搬运、计算和存储,不创造信任: + +```text + ① 发布密钥(CI) ──▶ 「跑什么代码」 已有:清单签名、seq 防重放 + ② 操作员密钥(管理员本人) ──▶ 「谁是舰队成员、去哪拨」 新增:Backend 持有不了 + ③ 节点自己的密钥 ──▶ 「我是我」 已有;Probe 要改成自己生成 + + 被攻破的 Backend 能做的:拒绝服务、看数据。 + 不能做的:加入新节点、改地址让 Probe 去探内网、塞恶意更新、冒充节点。 +``` + +## 5. 名册(roster) + +操作员签名的舰队成员表,节点本地固定操作员的公钥,只采用验签通过的名册。 + +```text + 管理端(管理员) Backend Agent / Probe + ┌──────────────┐ ①提交变更 ┌─────────┐ ┌──────────────┐ + │批准节点、改地址│───────────────▶│只存、转发│─② 名册 + 签名─▶ │本地固定操作员 │ + │或吊销 │◀─③ 挑战(哈希)─│ │ │公钥;验签通过 │ + │触控或确认 │ ④ 签名 │ │ │且 version 更大│ + │ │───────────────▶│ │ │才采用并持久化 │ + └──────────────┘ └─────────┘ └──────────────┘ +``` + +内容:`{version, expires_at?, 成员: [{id, 角色, 公钥(SPKI), Agent 的公网地址、端口和 TLS 指纹}]}`。 + +- Agent 只接受名册里的 Probe;Probe 只拨名册里的 Agent,并且只用名册里写的地址(缺口 G3、G6 因此消失,不需要另外的拨号白名单)。 +- **不可回放**:节点持久化见过的最大 `version`,低于它的名册一律拒绝,所以 Backend 无法拿旧名册复活已吊销的节点。 +- 可选的 `expires_at` 防止 Backend 截留新名册(冻结攻击)。 +- 每次改动成员或地址都要操作员签名一次:批准新节点(加入令牌的自动批准只让它连上,进名册仍要签)、公网地址变了、吊销。 +- 任务集、测试请求这类内容不进名册:它们由 Backend 计算,被节点本地硬上限兜住。 + +## 6. 操作员凭证 + +节点本地配置里固定一串操作员凭证,每条带类型,验签代码按类型分支;任一条通过即可,所以丢了一个设备不会被锁死,换凭证是改节点本地配置(重新运行部署脚本)。 + +| 类型 | 适合 | 节点怎么验 | 强度 | +| --- | --- | --- | --- | +| `webauthn` | 有通行密钥的设备 | 固定凭证公钥,验 `authenticatorData ‖ SHA256(clientDataJSON)` 的 ECDSA 签名,检查 `challenge` 等于名册哈希;注册时 attestation 选 `none`,信任浏览器实现,不验证设备证明 | 最强,每次改动需要本人触控 | +| `ecdsa` | 没有通行密钥的设备 | 管理端用 WebCrypto 生成不可导出的 P-256 密钥,存在受信任的 Chromium 里,直接对名册哈希签名 | 中:本机被入侵或页面被 XSS 时可被静默调用 | +| `oidc` | 完全不方便装密钥的设备 | 授权请求的 `nonce` 填名册哈希,节点验 ID Token 的签名(固定 IdP 的签名公钥)、`aud`、`sub`、`nonce` 和新鲜度 | 较弱:IdP(例如 Gitea)被攻破就等于操作员被攻破;节点没有 HTTP 客户端,IdP 公钥只能固定在本地,换密钥要改配置 | + +三种在节点上都只用 BCL 的 ECDSA 或 RSA 验签,不增加依赖和资源。 + +## 7. 各类下行消息怎么处置 + +| 下行内容 | 授权 | 被攻破的 Backend 最多能做什么 | +| --- | --- | --- | +| 舰队成员、对端公钥、Agent 地址 | **操作员签名**(名册) | 什么都改不了 | +| 更新清单 | 发布密钥(已有) | 不能伪造;能转发旧的合法清单,`seq` 防回滚 | +| 任务集、测试请求 | Backend 计算;节点本地硬上限兜底(已有) | 只能在名册成员之间、本地速率和端口上限内发测量流量 | +| 重启计划(`moshiro.schedule`) | Backend;节点本地启动后 10 分钟内不触发 | 最多让节点每 10 分钟重启一次 | +| 未来的 `extension`(exec 等) | **操作员签名的一次性能力令牌**,见第 11 节 | 什么都做不了 | + +另外 Backend 自己对节点也有一条应用层身份:Backend 有长期签名密钥,节点在注册时固定它,下行消息带签名、节点 ID、序号和有效期。它防的是你的 Cloudflare、frp 链路被劫持(G1);对 Backend 自己被攻破没有帮助,那一类靠名册。 + +## 8. 协议的方向与平面 + +### 8.1 三个平面,各认各的凭证 + +```text + 外部面(人、浏览器、脚本) /api/public(匿名只读,SSE) /api/admin(Cookie + CSRF,或 API 令牌) + /install.sh /openapi /healthz /metrics + 节点面(节点主动拨入) /api/agent/v1/enroll(加入令牌,一次性) /api/agent/v1/rpc(wss,节点签名认证) + 对等面(Probe 拨 Agent) TLS 控制 7443/tcp、数据 7443/udp、7444/tcp(名册里的成员才能通过) +``` + +- **每个平面只认自己的凭证**:节点面不接受令牌和 Cookie;外部面不接受节点签名;对等面只认名册成员。任何一个平面的凭证都换不到另一个平面的权限。 +- **连接方向固定**:节点面永远是节点拨入 Backend,Backend 从不主动连节点,所谓「下发」只是在节点拨入的连接上发下行消息;对等面永远是 Probe 拨 Agent,Agent 不拨 Probe。 +- **外部面不能直通节点面**:管理端的操作(例如发起测试)只会触发 Backend 里一个**强类型**的下行方法,管理端不能指定方法名,也不能把请求体原样转给节点。现状已满足(`IAgentGateway` 里全是强类型方法),要写成约束并用测试守住。 +- 现状:三个平面共用同一个监听端口和同一条 ASP.NET 管线,靠路径和凭证类型区分。目标是把节点面和外部面做成可以分别配置监听地址(默认仍同一个),这样管理面可以只对内网或 VPN 开放,节点面单独放出去。 + +### 8.2 每个方法声明方向、类型和角色 + +**同一个方法名只允许一个发起方**,不存在「双向方法」。需要两个方向交互时,定义两个方法。一个方法声明四件事: + +| 声明 | 取值 | 含义 | +| --- | --- | --- | +| 发起方 | `B→A`、`B→P`、`A→B`、`P→B`、`P→A`、`A→P` | 只有它能发出这个方法,对方收到就是违规 | +| 类型 | 请求(有响应)、通知(无响应) | 通知不允许回复;响应只对应一个未决的请求 ID,且那个方法必须是请求 | +| 角色 | `A`、`P`、`A+P` | 节点角色不对就拒绝(Probe 不能调 Agent 的方法) | +| 授权 | 无、`operator`(需要操作员签名的能力令牌) | 为 extension 预留,核心方法都是「无」 | + +方向总表(来自 [protocol.md](protocol.md),实现时以代码里的声明为准): + +| 方法 | 方向 | 类型 | 角色 | +| --- | --- | --- | --- | +| `fleet.challenge`、`fleet.state.sync`、`fleet.outbox.ack`、`fleet.revoked` | B→节点 | 见 Fleet 说明 | A+P | +| `fleet.auth`、`fleet.hello`、`fleet.heartbeat` | 节点→B | 见 Fleet 说明 | A+P | +| `moshiro.test.run`、`moshiro.test.cancel`、`moshiro.probe.disconnect` | B→A | 请求 | A | +| `moshiro.update.offer`、`moshiro.update.apply`、`moshiro.schedule` | B→节点 | 通知 | A+P | +| `moshiro.test.status`、`moshiro.event`、`moshiro.host` | A→B | 通知 | A | +| 二进制通道 16(`RawChunk`) | A→B | 发件箱 | A | +| 对等面的方法和通道 17(列块) | P→A、A→P,按 [protocol.md](protocol.md) | 按声明 | P、A | + +### 8.3 强制方式 + +1. **一份声明表,单一真源**,放在 `MoShiro.Protocol`(Fleet 自己的 `fleet.*` 放在 Fleet);文档里的方向表由它生成或对账。 +2. **发送前检查**:发错方向是编程错误,直接失败。 +3. **接收时检查**:收到不该由对方发起的方法、未声明的方法、角色不对的方法,丢弃并计数,连续违规断开连接。 +4. **响应匹配**:只接受对应未决请求 ID 的响应,不接受主动「响应」。 +5. **对账测试**(`ArchitectureTests`):声明表和实际注册的处理器逐项比对,新增方法绕不过声明。 +6. **下行消息只是建议,不是命令**:节点按本地硬上限执行或拒绝;任何带副作用的敏感能力都不走核心的下行消息,见第 11 节。 + +**已实现**(阶段零): + +- 声明表:`Fleet.Rpc/MethodTable.cs`(`MethodSpec`、`ChannelSpec`、`MethodTable`);框架的 `fleet.*` 在 `FleetMethodTable`,`moshiro.*` 和两条二进制通道在 `MoShiro.Protocol` 的 `MoShiroMethodTable`;角色是 `NodeRoles`(服务端按节点 ID 前缀判断)。声明表按(平面,名字)查找,同名方法在两个平面里可以有不同的发起方(`fleet.state.sync` 在节点面是 Backend→节点,在对等面是 Agent→Probe)。 +- 卡口:`RpcRouter` 创建时必须给出本侧(`RpcSide`)和声明表。**注册**处理函数时核对方向、类型、角色,不符就在启动时失败;**分发**时拒绝没有声明、方向不对、请求和通知用反了、角色不允许的消息,回 `fleet_protocol_violation` 并报告违规;**发送**前核对本侧有没有资格发起(`CheckSend`,违反是程序缺陷,抛异常)。 +- 违规处理:`RpcSession` 和服务端的 `NodeConnection` 对违规计数,超过上限(`MaxViolations`,默认 20)就断开连接;二进制通道也在服务端按声明核对(节点只能往声明过、并且允许它的角色使用的通道发)。 +- 对账测试:`Fleet.Tests/MethodDirectionTests` 覆盖卡口本身;`MoShiro.Backend.Tests/ProtocolDeclarationTests` 核对「每个方法常量都声明了」「文档里写了每个方法」「Backend、Agent、Probe 两个平面各自该处理的方法都注册了处理函数」「每条声明的通道都有收件箱」。 +- 这次落地时对账发现了两处不一致并修掉:文档漏了 `fleet.peer.challenge`、`fleet.peer.auth`;Agent 的 `moshiro.schedule` 处理函数原来只在传入重启控制器时才注册。 + +## 9. 节点自己被攻破时:数据可信与归属 + +### 9.1 测量数据的可信(G8、G9、G10) + +测量数据是这个项目的产出,不能只靠传输层可信:你的链路有 Cloudflare 这样的中间环节,Agent 又要转发 Probe 的数据。 + +- **产生者签名**:每个列块由 Probe 签名,每个原始块由上传它的节点签名;Agent 转发 Probe 的块时原样保留 Probe 的签名。签名覆盖块的摘要(见第 13 节),签名串带用途前缀 `chunk` 和节点 ID。Backend 入库时逐个验签,不通过的丢弃并计数。每 10 秒一块,一次 ECDSA 验签的开销可以忽略。 +- **归属校验**:入库时核对块里的 Probe 是名册里和这个 Agent 有线路的成员,不是就丢弃并计数;线路只由名册里的(Probe,Agent)对产生。 +- **合理性检查**:`t3−t2` 必须小于 `t4−t1`,各时间戳单调;违反的样本不计入指标,记为异常并产生事件。 +- 边界:这样中间环节改不了数据,被攻破的 Agent 改不了 Probe 一侧的数据;但它仍然可以谎报自己一侧的 `t2`、`t3`,合理性检查只能限制它的幅度,消除不了,因为这一侧的真值只有它自己知道。 + +### 9.2 对 Backend 的资源和输入 + +- **按节点的配额**:每个节点的字节、块数和速率上限,只是计数,不占资源;全局再限制同时解码的块数(单块解码上限是 64 MiB,多个节点同时发大块可以耗尽内存)。 +- **节点上报的字符串**:`hello` 里的 `platform`、事件的 `detail` 等一律当不可信数据,入库前统一校验长度、字符集和换行,不依赖前端转义;管理端加 CSP,不渲染成 HTML。 +- **匿名公开接口**:登录已经有限流,但热力图、分布这类重查询只有范围和指标数的上限,要加每源限速和结果缓存(TODO 里的人机验证会部分覆盖)。 +- **webhook**(TODO 里有):实现之前先定约束:只允许 HTTPS、禁止私网和回环地址、不跟随重定向、限制超时和响应大小、解析之后再检查地址。 +- 解码器保持大小上限;模糊测试已在 TODO。 + +### 9.3 对对端和身份 + +- 被攻破的 Probe 只能拿到名册授权的会话,Agent 的速率上限由本地兜底;被攻破的 Agent 发来的数据只会影响它自己的线路。 +- **身份被偷**:操作员签一份新名册吊销它,所有节点立即拒绝,且因为 `version` 递增不可回放。 +- **Probe 私钥要自己生成**:现在 Backend 知道它(G2),名册也防不了这一点。改成和 Agent 一样用加入令牌注册,私钥只在节点上。 +- **密钥按用途拆开**(G11):面向 Backend 的和面向对等面的是两把独立的 P-256 密钥;敲门、会话等对称用途用 KDF 加用途标签派生,不复用签名密钥直接做别的事。 + +### 9.4 对设计的补充要求(实现时核实现状) + +- **所有下行状态文档都防回放**:节点持久化每种状态文档见过的最大 `version`,低版本一律拒绝,不只是名册。 +- **UDP 会话迁移要做路径验证**:协议允许会话在 NAT 映射变化后跟随新的源地址,路径上的攻击者可以借此把会话牵到自己的地址。迁移前先做一次挑战应答,验证前限制发送量(不超过收到字节数的 3 倍),类似 QUIC 的 path validation。 +- **名册里的主机名解析之后再检查地址**:签名保护的是字符串,不是解析结果;默认只放行公网地址,内网需要节点本地显式允许;名册里优先写地址字面量。 +- **不协商、不降级**:现在固定 ECDSA P-256 和固定的签名串格式,没有算法协商,保持这一点;升级是硬失败,任何「兼容旧版」的开关都要有明确的下线期限。 + +### 9.5 默认值 + +Agent 的 `max_total_mbps` 默认是 1200,对 1 核 1 GB 的小机器没有意义,却是被攻破的 Backend 能让 Agent 向某个 Probe 打出去的上限。默认值应该保守,需要的人再调高;`allowed_target_cidrs` 默认为空是对的。 + +## 10. 监听隐身:预握手敲门 + +目标:Agent 的 TCP 端口对没有完成挑战的人**不写出任何字节**,包括 TLS 的 ServerHello,扫描器拿不到证书和 TLS 指纹(G4)。UDP 端口本来就是静默的。 + +```text + Probe Agent TCP 7443 / 7444 + │── TCP 连接 ───────────────────────────────────▶ accept(内核已完成三次握手) + │── 敲门包 32 字节:nonce(16) │ tag(16) ─────────▶ 读 32 字节(限时 3 秒) + │ tag = HMAC(K_pair, 时间窗 ‖ nonce) 用名册里每个 Probe 的 K_pair 逐个核对 + │ ├ 合法:继续 TLS 握手 → 现有的签名认证 + │◀──────────── TLS ServerHello ─────────────────── └ 不合法或超时:什么都不发,直接断开 +``` + +- `K_pair = HKDF(ECDH(Probe 身份私钥, Agent 身份公钥), "moshiro-knock-v1")`:只有名册里的成员算得出;ECDH 和签名共用同一把 P-256 密钥时,用不同的 HKDF 标签做域隔离。 +- 时间窗 30 秒,接受前后各一个窗(容忍时钟偏差);`nonce` 在窗内去重,防重放;每个来源地址限速,避免陌生人用随机数据消耗 HMAC 计算。 +- 这是破坏性变更:旧 Probe 不发敲门包,会被拒绝。 + +**局限,必须说清楚**:TCP 的 SYN-ACK 由内核发出,用户态进程拦不住,所以端口在扫描器眼里仍是「开着,但不说话」,**不是「没监听」**。要做到真正的没监听,有两条路: + +- **等级二:UDP 单包授权**。TCP 平时不监听;Probe 先发一个签名的 UDP 敲门包(UDP 对陌生包本来就静默),Agent 才临时开 TCP 监听几秒,并只放行敲门者的源地址。代价是每次连接多一个往返、要处理 UDP 丢包重试、多出口的 Probe 源地址可能对不上。 +- 宿主防火墙规则:需要 root,不适合小鸡节点,不做。 + +先做等级一,等级二作为可选。 + +## 11. extension:敏感能力的归宿 + +Agent 和 Probe 的核心永远是纯采集器。**所有敏感能力**(exec、web ssh 和终端、文件读写、抓包等)都不进核心,做成 extension,**用户可选安装,默认没有**。这样核心保持 NativeAOT、体积小、没有可被滥用的能力,需要这些功能的人装上,不需要的人什么都不用管,便利性不丢。 + +```text + 节点主机 + ┌ 核心容器(NativeAOT,纯采集)┐ ┌ extension 容器(可选,例如 ssh)┐ + │ 身份 A:节点自己的密钥 │ │ 身份 E:自己的密钥,独立注册 │ + │ 方法:moshiro.* │ │ 方法:ext.ssh.* │ + └───────────────┬──────────────┘ └───────────────┬────────────────┘ + │ wss(节点面) │ wss(节点面,另一条独立的会话) + ▼ ▼ + Backend(核心不知道 extension,extension 不能调用核心) +``` + +- **独立进程,不是插件**:extension 是单独的二进制和镜像,和核心之间**没有进程内接口**,核心不加载任何外部代码、不用反射和动态加载,所以核心的 NativeAOT 和「不能 exec」都不受影响。extension 自己用什么技术实现都行(可以是 NativeAOT,也可以不是)。 +- **独立身份**:extension 有自己的密钥、自己向 Backend 注册,在名册里是单独的成员(角色 `extension`,记录它依附的节点 ID)。核心被攻破碰不到 extension 的密钥,反过来也一样。 +- **独立发布**:沿用签名更新的流程,`component` 不同(例如 `ext-ssh`),独立的镜像、独立的清单;装不装、升不升级都不影响核心的更新。 +- **独立命名空间**:extension 的方法只能是 `ext.<名字>.*`,同样进方向声明表;核心的声明表里没有 `ext.*`,核心收到一律拒绝。extension 只响应自己命名空间的方法。 +- **安装像装一个选项**:部署脚本加一个选项(例如 `--extension ssh`)拉起对应容器并写好配置;管理端在节点详情里出现对应入口。卸载就是停掉那个容器。 +- **本地显式启用**:extension 的本地配置里有 `enabled`、允许的操作员凭证、允许的命令集或目标范围(例如允许登录的用户、允许访问的路径),默认关闭,没有配置就什么都不做。 +- **每次调用都要授权**:操作员签名的一次性能力令牌,`{节点 ID, extension, 命令哈希, not_before, not_after, nonce}`,有效期以秒计;extension 用本地固定的操作员凭证验签。被攻破的 Backend 不能启用 extension,也伪造不了一次调用。 +- **交互式会话的内容对 Backend 不透明**(例如 web ssh):浏览器和 extension 在能力令牌携带的公钥基础上协商一次性会话密钥,Backend 只转发密文,不应该能读到终端里的内容。具体做法还没定,见第 14 节。 +- **不做的**:核心里的任何 exec、文件写、加载 extension;用 extension 绕过核心的本地硬上限(extension 有它自己的上限,核心的不放宽)。 + +## 12. 实施顺序 + +```text + 零 协议方向声明表 + 收发校验 + 对账测试 G7,第 8 节;同时把 NodeConnection 按职责拆开(第 13.2 节) 已完成,不破坏 + 一 Probe 自己生成密钥 + 加入令牌注册 + 密钥按用途拆开 G2、G11 破坏性:旧 probe.json 失效 + 二 预握手敲门(等级一) G4;先基于现有的对端公钥列表,名册出来后换成名册 破坏性:旧 Probe 被拒绝 + 三 操作员凭证 + 名册 + 节点验签与持久化 G1 的大部分、G3、G6;webauthn 和 ecdsa 先做,oidc 随后 破坏性:节点要固定操作员公钥并重新注册 + 四 测量数据的可信 + 入库配额和输入校验 + CSP G5、G8、G9、G10、T5(第 9.1、9.2 节) 破坏性:块格式加签名 + 五 Backend 应用层身份固定 + 下行消息签名 G1 的剩余部分(链路劫持) 破坏性 + 六 extension 框架 独立身份和名册条目、`ext.*` 命名空间、能力令牌、部署脚本的 --extension 选项(先不做具体的 ssh) + 可选 等级二:UDP 单包授权;UDP 会话迁移的路径验证 +``` + +阶段一和二是敲门的前提(双方身份密钥都由节点自己生成)。项目允许破坏性变更,当前只有三台节点,重新注册的成本很小。 + +## 13. 密码学原语与代码落点 + +### 13.1 原语现状,以及为什么不用 BLAKE3 + +| 用途 | 现状 | +| --- | --- | +| 身份签名 | ECDSA P-256 加 SHA-256(BCL) | +| 数据面逐包认证 | HMAC-SHA256 截断到 16 字节(`UdpPacket`、`TcpWire`) | +| 会话密钥派生 | HKDF-SHA256(`SessionKeys`) | +| 令牌哈希、SPKI 指纹 | SHA-256 | +| 原始块完整性 | CRC32C,不是密码学校验 | + +**BLAKE3 的评估(已定:不采用)**: + +- 适合的位置:数据面的逐包 MAC(原生的 keyed 模式,一个原语、32 字节密钥)、KDF(`derive_key` 自带用途标签,比手拼 HKDF 的 `info` 更不容易出错)、给块做摘要再签名(第 9.1 节需要一个块摘要)。 +- 不能替换的位置:ECDSA P-256 本身用 SHA-256,令牌哈希和 SPKI 指纹也继续用 SHA-256。所以 BLAKE3 是**多一个**哈希族,不是换掉一个,两套原语的表面都要维护。 +- 代价:BCL 里没有;Probe 和 Agent 不引用 NuGet 包,只能在仓库里自己实现,几百行的便携实现加官方测试向量,这是要长期维护的密码学代码。托管实现没有 SHA-NI 加速,而 .NET 的 SHA-256 走系统 OpenSSL、有硬件加速,所以在 x86 上未必更快;收益主要在没有 SHA 指令的 ARM、MIPS 路由器上。 +- 结论:**不采用,原语保持现状**(签名 ECDSA P-256 加 SHA-256,逐包 MAC 用 HMAC-SHA256,KDF 用 HKDF-SHA256):不增加一套要长期维护的自实现密码学代码,也不多一个哈希族。块摘要(第 9.1 节)用 SHA-256。若以后在目标硬件上实测发现逐包 MAC 是瓶颈,再单独评估,届时只替换 `Fleet.Identity` 里 `Mac` 和 `Kdf` 两个静态函数,调用方不动。 + +### 13.2 代码落点与分层原则 + +目标是**每个机制只有一个落点,层数尽量少,方便维护**: + +1. **一个机制,一个落点,纯函数优先**:输入字节、输出判定,不引入新的接口、适配层、端口或注册。节点侧继续不用 DI。 +2. **声明和校验放在同一个项目**:`fleet.*` 的声明表在 `Fleet.Rpc`,`moshiro.*` 的在 `MoShiro.Protocol`;校验统一在 `RpcRouter`,节点和服务端共用,这是唯一的卡口。 +3. **新代码进现有项目,不新增项目**:Fleet 侧进 `Identity`、`Rpc`、`Link`;Backend 侧的入库检查就放在 `IngestService`,不为安全另建一层。 +4. **先收缩再加**:`NodeConnection`(703 行)里混着认证、收包分发、状态推送、在线状态和帧编解码,先按职责拆成同一项目里的几个文件,再往里加方向校验;否则每加一个安全机制它就更难读。**已拆**:`NodeConnection.cs`(生命周期)、`.Auth`、`.Receive`、`.Send`、`.State`、`.Wire` 六个 partial 文件,原来的 729 行主文件现在 143 行。 + +| 机制 | 落点 | 调用点 | +| --- | --- | --- | +| 方向和角色声明、收发校验 | 声明:`Fleet.Rpc`、`MoShiro.Protocol`;校验:`RpcRouter` | 节点和服务端各一处创建路由器的地方 | +| 对账测试 | `ArchitectureTests`(已有) | 比对声明表和实际注册的处理器 | +| 敲门的构造与验证 | `Fleet.Identity` 里一个静态类 | 服务端两处:`Fleet.Link.PeerListener`(7443)、`Measurement.TcpAcceptor`(7444);客户端两处:`PeerConnector`、`TcpLink` | +| 操作员凭证验签、名册 | `Fleet.Identity`(三种凭证各一个验签函数) | `Fleet.Link` 的 `StateHandler` 采用名册,用 `ConfigStore` 持久化版本 | +| 块和列块的签名与验签 | `MoShiro.Protocol/Raw`(格式和签名都在这里) | 签名:Probe 的列块、节点上传的原始块;验签:`IngestService` | +| 归属校验 | `IngestService` | 入库的唯一入口 | +| 合理性检查 | `Backend.Analysis` 的 `EchoOps` | 计算指标的入口 | +| 密钥按用途拆开、`Mac`、`Kdf` | `Fleet.Identity` | 原有的使用方 | + +## 14. 还没定的事 + +1. 操作员凭证第一批做 `webauthn` 和 `ecdsa`,`oidc` 放在名册稳定之后,是否这样;还是 `oidc` 一起做。 +2. 敲门先做等级一,等级二先记 TODO,是否这样。 +3. 严格程度:节点没有固定操作员公钥就拒绝一切 Backend 给的对端和地址(推荐),还是保留一个过渡期兼容旧行为。 +4. 管理员域(T4)的审计日志、令牌过期、危险操作二次确认:还没有核对现状,也没有设计。 +5. 节点面和外部面是否拆成两个监听端口(8.1 的目标):拆的话部署文档、Caddy 和隧道的配置都要跟着改。 +6. extension 的交互式会话要不要做到「Backend 只转发密文」(端到端加密),还是先做成 Backend 可见明文、以后再加。 +7. extension 的第一个具体实现做不做 web ssh,放在哪个阶段。 +8. 块摘要固定 SHA-256,不预留算法字段(不协商、不降级,见第 9.4 节)。 +9. Agent 的 `max_total_mbps` 默认值调到多少(第 9.5 节),比如 100。 +10. 块的签名放进格式的哪个位置、是否影响现有 `MSRC` 容器版本(第 9.1 节)。