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 节)。