feat: pinned agents and probes only accept peers carrying the operator's proof; the panel ships the signed operator chain with the rosters [CI SKIP]

This commit is contained in:
justme committed 2026-10-07 23:24:33 +08:00
1 parent fc2bbdcc6d
commit 9bb6841dce
11 files changed
+129 -12

No files matched your search

+1 -1
View File
@@ -81,7 +81,7 @@ public sealed class AgentApp : IDisposable
};
var chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log);
var targets = new TargetRunners(chunks, identity.NodeId, _clock, time, log);
_tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _udp.BoundPorts, () => _tcp.BoundPorts);
_tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _udp.BoundPorts, () => _tcp.BoundPorts, TrustAnchor.Load(config.Fleet.IdentityDir, log));
_hub = new PeerHub(_sessions, _clock, chunks, _tasks, _uplink, time, log);
_tests = new TestRunner(_tasks, _sessions, _streams, env, _uplink, config.Limits.MaxConcurrentTests, transport => transport == "udp" ? _udp.BoundPorts : _tcp.BoundPorts, time, log);
@@ -1,4 +1,6 @@
using System.Text.Json;
using Fleet.Hosting;
using Fleet.Identity;
using Fleet.Core;
using Fleet.Rpc;
using MoShiro.Measurement;
@@ -25,14 +27,16 @@ public sealed class TaskManager
private readonly TargetRunners _targets;
private readonly Func<IReadOnlyList<int>> _udpPorts;
private readonly Func<IReadOnlyList<int>> _tcpPorts;
private readonly TrustAnchor _trust;
private readonly Dictionary<ProbeSession, Dictionary<string, LineEntry>> _sessionLines = [];
private volatile Dictionary<string, byte[]> _peers = new(StringComparer.Ordinal);
private Dictionary<string, (TaskSpec Task, TaskDecision Decision)> _lineSpecs = [];
private long _version;
private BudgetGuard? _guard;
public TaskManager(Func<BudgetGuard> guardFactory, StreamRegistry streams, BlockEnv env, TargetRunners targets, Func<IReadOnlyList<int>> udpPorts, Func<IReadOnlyList<int>> tcpPorts)
public TaskManager(Func<BudgetGuard> guardFactory, StreamRegistry streams, BlockEnv env, TargetRunners targets, Func<IReadOnlyList<int>> udpPorts, Func<IReadOnlyList<int>> tcpPorts, TrustAnchor? trust = null)
{
_trust = trust ?? TrustAnchor.None;
_guardFactory = guardFactory;
_streams = streams;
_env = env;
@@ -98,12 +102,22 @@ public sealed class TaskManager
_ = ct;
var result = new StateSyncResult { AppliedVersion = version };
// 固定了操作员创世的节点:先验面板带来的操作员集合链(从固定的创世出发),再要求每个 Probe 带着操作员批准的证明。
// 面板自己改名册、塞进自己的 Probe 公钥是过不了这一关的:它签不出操作员的委托。
_trust.Update(set.TrustChain);
var peers = new Dictionary<string, byte[]>(StringComparer.Ordinal);
foreach (PeerSpec peer in set.Peers)
{
try
{
peers[peer.ProbeId.Value] = Convert.FromBase64String(peer.PublicKey);
byte[] key = Convert.FromBase64String(peer.PublicKey);
if (_trust.VerifyPeer(peer.Proof, key, "probe") != TrustFailure.None)
{
result.Rejected.Add(new RejectedItem { Id = peer.ProbeId.Value, Reason = RejectReasons.PolicyDenied });
continue;
}
peers[peer.ProbeId.Value] = key;
}
catch (FormatException)
{
@@ -1,5 +1,6 @@
using System.Text.Json;
using Fleet.Core;
using Fleet.Identity;
using Fleet.Hosting;
using MoShiro.Protocol;
@@ -16,11 +17,13 @@ public sealed class AgentDirectory
private readonly object _gate = new();
private readonly ConfigStore _store;
private readonly INodeLog _log;
private readonly TrustAnchor _trust;
private List<KnownAgent> _agents = [];
private long _version;
public AgentDirectory(ConfigStore store, string configDirectory, INodeLog log)
public AgentDirectory(ConfigStore store, string configDirectory, INodeLog log, TrustAnchor? trust = null)
{
_trust = trust ?? TrustAnchor.None;
_store = store;
_log = log;
string cache = Path.Combine(configDirectory, FileName);
@@ -64,12 +67,34 @@ public sealed class AgentDirectory
}
}
private static bool TryKey(string text, out byte[] key)
{
try
{
key = Convert.FromBase64String(text);
return key.Length > 0;
}
catch (FormatException)
{
key = [];
return false;
}
}
/// <summary>应用面板下发的列表。版本不比当前新的忽略(乱序到达的旧推送)。</summary>
public void Apply(ProbeAgentList result)
{
var next = new List<KnownAgent>();
_trust.Update(result.TrustChain);
foreach (AgentEndpoint agent in result.Agents)
{
// 固定了操作员创世的 Probe 只连操作员批准过的 Agent(证明验不过的,面板塞进来的也不连)
if (!TryKey(agent.PublicKey, out byte[] key) || _trust.VerifyPeer(agent.Proof, key, "agent") != TrustFailure.None)
{
_log.Warn("an agent from the panel carries no valid operator proof; not connecting", ("agent", agent.AgentId.Value));
continue;
}
if (!string.IsNullOrWhiteSpace(agent.Host))
{
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Host, TcpPort = agent.TcpPort, PublicKey = agent.PublicKey });
+1 -1
View File
@@ -27,7 +27,7 @@ public sealed class ProbeApp : IDisposable
public ProbeApp(ProbeFile file, LoadedNodeIdentity identity, string configDirectory, TimeProvider time, INodeLog log, RestartController? restart = null)
{
_log = log;
var directory = new AgentDirectory(new ConfigStore(configDirectory), configDirectory, log);
var directory = new AgentDirectory(new ConfigStore(configDirectory), configDirectory, log, TrustAnchor.Load(file.Fleet.IdentityDir, log));
_env = new ConnectionEnv(identity.PeerIdentity, identity.NodeId.Value, file.Limits, directory, new MonoClock(time), time, log);
directory.Changed += () => _changed.Release();
@@ -152,6 +152,10 @@ public sealed class AgentEndpoint
/// <summary>Agent 面向 Probe 的 TLS 证书公钥(SPKI,base64):证书按它的 SHA-256 指纹固定,敲门密钥由它和 Probe 的身份密钥派生。</summary>
[JsonPropertyName("public_key")]
public string PublicKey { get; set; } = "";
/// <summary>操作员批准这个 Agent 的证明;固定了创世的 Probe 没有它就不连。</summary>
[JsonPropertyName("proof")]
public MemberProofDto? Proof { get; set; }
}
/// <summary>期望状态文档(Fleet state type = <c>probe_agents</c>):Backend 直接下发给每个 Probe 的 Agent 列表。</summary>
@@ -164,6 +168,9 @@ public sealed class ProbeAgentList
[JsonPropertyName("agents")]
public List<AgentEndpoint> Agents { get; set; } = [];
[JsonPropertyName("trust_chain")]
public List<SignedStatementDto>? TrustChain { get; set; }
}
/// <summary>注册元数据(MoShiro 定义,经 Fleet 的 <c>meta</c> 字段带上来)。</summary>
@@ -40,6 +40,10 @@ public sealed class PeerSpec
/// <summary>Probe 的身份公钥(SPKI,base64)。</summary>
[JsonPropertyName("pubkey")]
public string PublicKey { get; set; } = "";
/// <summary>操作员批准这个 Probe 的证明(passkey 签的委托加码的私钥签的绑定)。固定了创世的 Agent 没有它就不接纳。</summary>
[JsonPropertyName("proof")]
public MemberProofDto? Proof { get; set; }
}
/// <summary>期望状态文档(Fleet state type = <c>task_set</c>):Backend 为每个 Agent 计算的有效任务集。</summary>
@@ -56,6 +60,10 @@ public sealed class TaskSet
/// <summary>允许连接到这个 Agent 的 Probe 公钥。</summary>
[JsonPropertyName("peers")]
public List<PeerSpec> Peers { get; set; } = [];
/// <summary>操作员集合的链(passkey 签的,创世在前):节点从固定的创世指纹验过才采用。</summary>
[JsonPropertyName("trust_chain")]
public List<SignedStatementDto>? TrustChain { get; set; }
}
public static class TaskSchedule
@@ -40,7 +40,7 @@ public static class TrustEndpoints
return TypedResults.Json(Mappers.ToDto(state), ApiJsonContext.Default.TrustStateDto);
}
private static async Task<IResult> CommitAsync(HttpContext http, SignedStatementDto signed, TrustService trust, BackendOptions options)
private static async Task<IResult> CommitAsync(HttpContext http, SignedStatementDto signed, TrustService trust, AgentService agents, BackendOptions options)
{
PasskeyContext context = AuthEndpoints.PasskeyContextOf(http, options);
// 第一次(创世)是整个信任根的起点:除了 passkey 签名,还要凭证窗口,避免只拿到一个密码会话的人抢先立根
@@ -55,6 +55,7 @@ public static class TrustEndpoints
return Problems.From(http, error);
}
await agents.AfterAgentSetChangedAsync(http.RequestAborted); // 新的操作员集合要随名册下发给节点
http.SetDetail($"operator set v{state.Version}");
return TypedResults.Json(Mappers.ToDto(state), ApiJsonContext.Default.TrustStateDto, statusCode: StatusCodes.Status201Created);
}
@@ -24,9 +24,11 @@ public sealed partial class AgentService
private readonly NodeInfoService _nodes;
private readonly LinkService _links;
private readonly TrustService _trust;
public AgentService(IAgentStore agents, AgentRuntime runtime, IAgentGateway gateway, TaskPlanner planner, ISettingsStore settings, NodeInfoService nodes, LinkService links)
public AgentService(IAgentStore agents, AgentRuntime runtime, IAgentGateway gateway, TaskPlanner planner, ISettingsStore settings, NodeInfoService nodes, LinkService links, TrustService trust)
{
_trust = trust;
_links = links;
_nodes = nodes;
_agents = agents;
@@ -169,9 +171,13 @@ public sealed partial class AgentService
TcpPort = agent.ProbeTcpPort,
UdpPorts = [.. agent.ProbeUdpPorts],
PublicKey = Convert.ToBase64String(agent.TlsPublicKey!),
Proof = agent.Proof,
});
}
IReadOnlyList<Fleet.Core.SignedStatementDto> chain = await _trust.ChainAsync(ct);
result.TrustChain = chain.Count == 0 ? null : [.. chain];
return result;
}
@@ -21,10 +21,12 @@ public sealed class TaskPlanner : IDisposable
private readonly IProbeStore _probes;
private readonly ISettingsStore _settings;
private readonly LinkService _links;
private readonly TrustService _trust;
private readonly SemaphoreSlim _gate = new(1, 1);
public TaskPlanner(ITaskStore tasks, IAgentStore agents, IProbeStore probes, ISettingsStore settings, LinkService links)
public TaskPlanner(ITaskStore tasks, IAgentStore agents, IProbeStore probes, ISettingsStore settings, LinkService links, TrustService trust)
{
_trust = trust;
_links = links;
_tasks = tasks;
_agents = agents;
@@ -62,9 +64,12 @@ public sealed class TaskPlanner : IDisposable
foreach (ProbeRecord probe in probes.Where(p => p.RevokedAtMs is null && p.PeerPublicKey is { Length: > 0 } && links.IsLinked(p.Id, agentId.Value)))
{
document.Peers.Add(new PeerSpec { ProbeId = new ProbeId(probe.Id), PublicKey = Convert.ToBase64String(probe.PeerPublicKey!) });
document.Peers.Add(new PeerSpec { ProbeId = new ProbeId(probe.Id), PublicKey = Convert.ToBase64String(probe.PeerPublicKey!), Proof = probe.Proof });
}
IReadOnlyList<Fleet.Core.SignedStatementDto> chain = await _trust.ChainAsync(ct);
document.TrustChain = chain.Count == 0 ? null : [.. chain];
string hash = Convert.ToHexStringLower(SHA256.HashData(JsonSerializer.SerializeToUtf8Bytes(document, MoShiroJsonContext.Default.TaskSet)));
await _gate.WaitAsync(ct);
@@ -41,12 +41,14 @@ public sealed class AgentHarness : IAsyncDisposable
public byte[] TlsPublicKey => _identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo();
public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? configure = null, string? name = null)
/// <param name="signed">操作员签过委托的加入码(认证串、种子、创世指纹):节点按固定了创世的方式注册,名册里的 Probe 必须带操作员批准的证明。</param>
public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? configure = null, string? name = null, (string Auth, string Seed, string Genesis)? signed = null)
{
string directory = Path.Combine(Path.GetTempPath(), "moshiro-agent-" + Guid.NewGuid().ToString("N"));
string token = await backend.CreateJoinTokenAsync();
string token = signed?.Auth ?? await backend.CreateJoinTokenAsync();
using var http = new HttpClient();
Result<EnrollmentResult> enrolled = await EnrollmentClient.EnrollAsync(http, backend.BaseAddress, token, name ?? backend.UniqueName("agent"), directory, AgentApp.Version, null, TimeProvider.System, CancellationToken.None);
Result<EnrollmentResult> enrolled = await EnrollmentClient.EnrollAsync(
http, backend.BaseAddress, token, name ?? backend.UniqueName("agent"), directory, AgentApp.Version, null, TimeProvider.System, CancellationToken.None, signed?.Seed, signed?.Genesis, "agent");
Assert.True(enrolled.IsSuccess, enrolled.IsSuccess ? "" : enrolled.Error.ToString());
// 测试里的 Agent 只在回环上监听:手填 public_host(手动的优先于自动识别,Agent 上报本机真实地址也不会覆盖)
@@ -217,4 +217,53 @@ public sealed class TrustTests
using HttpResponseMessage anonymous = await _backend.Anonymous.PostAsync("/api/admin/trust", BackendFixture.Json(unsigned));
Assert.Equal(HttpStatusCode.Unauthorized, anonymous.StatusCode);
}
[Fact]
public async Task APinnedAgent_OnlyAcceptsProbesTheOperatorApproved_EvenIfThePanelListsOthers()
{
Operator op = await OperatorAsync();
(string agentSeed, string agentAuth, _) = await CreateSignedCodeAsync(op, role: "agent");
await using AgentHarness agent = await AgentHarness.StartAsync(_backend, signed: (agentAuth, agentSeed, op.Genesis));
// 面板里有一个没有操作员证明的 Probe(相当于面板自己塞进名册的),再有一个操作员批准的
using EnrolledProbe plain = await _backend.CreateProbeAsync();
(string seed, string auth, _) = await CreateSignedCodeAsync(op, role: "probe");
string directory = Path.Combine(Path.GetTempPath(), "moshiro-trust-id-" + Guid.NewGuid().ToString("N"));
Result<EnrollmentResult> enrolled = await EnrollAsync(auth, seed, op.Genesis, directory: directory);
Assert.True(enrolled.IsSuccess, enrolled.IsSuccess ? "" : enrolled.Error.ToString());
using LoadedNodeIdentity approved = LoadedNodeIdentity.Load(directory);
await BackendFixture.EventuallyAsync(
() => Task.FromResult(agent.App.IsPeerKnown(enrolled.Value.NodeId, Convert.ToBase64String(approved.PeerIdentity.PublicKeySpki))),
because: "the agent should accept the probe the operator approved");
Assert.False(agent.App.IsPeerKnown(plain.ProbeId, Convert.ToBase64String(plain.Identity.PeerIdentity.PublicKeySpki)), "a probe without the operator's proof must not be accepted");
}
[Fact]
public async Task APinnedProbe_OnlyConnectsToAgentsTheOperatorApproved()
{
Operator op = await OperatorAsync();
(string probeSeed, string probeAuth, _) = await CreateSignedCodeAsync(op, role: "probe");
string probeDirectory = Path.Combine(Path.GetTempPath(), "moshiro-trust-id-" + Guid.NewGuid().ToString("N"));
Assert.True((await EnrollAsync(probeAuth, probeSeed, op.Genesis, directory: probeDirectory)).IsSuccess);
(string agentSeed, string agentAuth, _) = await CreateSignedCodeAsync(op, role: "agent");
await using AgentHarness agent = await AgentHarness.StartAsync(_backend, signed: (agentAuth, agentSeed, op.Genesis));
MemberProofDto? proof = TrustAnchor.Load(agent.DataDirectory, NullNodeLog.Instance).OwnProof;
string key = Convert.ToBase64String(agent.TlsPublicKey);
string configDirectory = Path.Combine(Path.GetTempPath(), "moshiro-probe-cfg-" + Guid.NewGuid().ToString("N"));
var directory = new MoShiro.Probe.AgentDirectory(new ConfigStore(configDirectory), configDirectory, NullNodeLog.Instance, TrustAnchor.Load(probeDirectory, NullNodeLog.Instance));
directory.Apply(new MoShiro.Protocol.ProbeAgentList
{
Version = 1,
Agents =
[
new MoShiro.Protocol.AgentEndpoint { AgentId = new NodeId("agt_approved"), Host = "127.0.0.1", TcpPort = 1, PublicKey = key, Proof = proof },
new MoShiro.Protocol.AgentEndpoint { AgentId = new NodeId("agt_planted"), Host = "127.0.0.1", TcpPort = 2, PublicKey = key }, // 面板塞进来的:没有证明
new MoShiro.Protocol.AgentEndpoint { AgentId = new NodeId("agt_swapped"), Host = "127.0.0.1", TcpPort = 3, PublicKey = Convert.ToBase64String(RandomNumberGenerator.GetBytes(91)), Proof = proof }, // 拿着别人的证明、换了自己的公钥
],
});
Assert.Equal(["agt_approved"], directory.Agents.Select(a => a.AgentId));
}
}