justme
904ea2b0ae
ci/woodpecker/push/main-amd64-build Pipeline was successful
The hash a node reports about itself cannot be trusted (a swapped binary can copy the old hash), so the runtime binding only caught files that were replaced by something that did not lie. Provenance stays where it can be verified: the signed release manifest (digest) and the signed SHA256SUMS. - Remove BinaryBindingService, the node_bindings table (schema v15 drops it), BinaryPolicy, the handshake hash/manifest fields and the admin binary column. - Drop the per-platform "binaries" list from the release manifest, the hash scripts and the updater's installed.manifest copy. - ReleaseService.PublishAsync now verifies the release signature, so the admin POST /releases route is no longer an unsigned path; tests sign with a test key. - Docs: section 16 explains why the runtime check is gone.
352 lines
15 KiB
C#
352 lines
15 KiB
C#
using Fleet.Identity;
|
||
using System.Net;
|
||
using System.Text.Json;
|
||
using Fleet.Core;
|
||
using Fleet.Hosting;
|
||
using Fleet.Link;
|
||
using Fleet.Rpc;
|
||
using MoShiro.Measurement;
|
||
using MoShiro.Protocol;
|
||
|
||
namespace MoShiro.Agent;
|
||
|
||
/// <summary>Agent 的组装点：数据面、点对点监听、与 Backend 的链路都在这里接起来。没有全局状态，测试可以在同一进程里启动多个。</summary>
|
||
public sealed class AgentApp : IDisposable
|
||
{
|
||
public const string Version = MoShiro.Protocol.BuildInfo.Version;
|
||
|
||
private readonly AgentConfig _config;
|
||
private readonly bool _ipv6Supported;
|
||
private readonly KnockGate _knock;
|
||
private readonly TrustAnchor _trust;
|
||
private readonly RestartController _restart;
|
||
private readonly RestartController? _ownedRestart;
|
||
private readonly LoadedNodeIdentity _identity;
|
||
private readonly TimeProvider _time;
|
||
private readonly INodeLog _log;
|
||
private readonly MonoClock _clock;
|
||
private readonly AgentCounters _counters = new();
|
||
private readonly SessionRegistry _sessions = new();
|
||
private readonly StreamRegistry _streams = new();
|
||
private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
private readonly AgentAdmission _admission;
|
||
private readonly Outbox _outbox = new(64 << 20);
|
||
private readonly UpdateBridge _update;
|
||
private readonly BackendUplink _uplink;
|
||
private readonly UdpMux _udp;
|
||
private readonly TcpAcceptor _tcp;
|
||
private readonly PeerListener _peers;
|
||
private readonly PeerHub _hub;
|
||
private readonly TaskManager _tasks;
|
||
private readonly TestRunner _tests;
|
||
private readonly FleetLink _link;
|
||
|
||
/// <param name="dataDirectory">配置目录（容器里的 <c>/etc/moshiro</c>）：宿主机更新器和节点在这里交换清单、请求和状态。</param>
|
||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log, RestartController? restart = null)
|
||
{
|
||
_ownedRestart = restart is null ? new RestartController() : null;
|
||
_restart = restart ?? _ownedRestart!;
|
||
_update = new UpdateBridge(dataDirectory, new ConfigStore(dataDirectory).Write, log);
|
||
_config = config;
|
||
_identity = identity;
|
||
_time = time;
|
||
_log = log;
|
||
_clock = new MonoClock(time);
|
||
_uplink = new BackendUplink(time);
|
||
|
||
IPAddress bind = IPAddress.Parse(config.Listen.Bind);
|
||
if (bind.Equals(IPAddress.IPv6Any) && !System.Net.Sockets.Socket.OSSupportsIPv6)
|
||
{
|
||
bind = IPAddress.Any;
|
||
}
|
||
|
||
_ipv6Supported = bind.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6;
|
||
_admission = new AgentAdmission(_streams);
|
||
_udp = new UdpMux(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.UdpPorts);
|
||
// 预握手敲门：控制端口和数据端口共用一个关卡（限速和去重一起算）；名册就是期望状态里的 peers
|
||
_trust = TrustAnchor.Load(config.Fleet.IdentityDir, log);
|
||
_knock = new KnockGate(identity.PeerIdentity, RosterKeys, time);
|
||
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data, _knock);
|
||
|
||
// 监听方积木的环境：和 Probe 的拨号方共用同一套积木，区别只在这里装配的部分（准入策略、TCP 监听）
|
||
var env = new BlockEnv
|
||
{
|
||
Clock = _clock,
|
||
Log = log,
|
||
Session = () => null, // 每个 Probe 会话各用自己绑定的环境（BlockEnv.WithSession）
|
||
Udp = _udp,
|
||
TcpListen = _tcp,
|
||
Admission = _admission,
|
||
Counters = _counters.Data,
|
||
MaxPacketsPerSecond = config.Limits.MaxTotalPps,
|
||
MaxBulkMbps = config.Limits.MaxTotalMbps,
|
||
};
|
||
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, _trust);
|
||
_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);
|
||
|
||
_tasks.OnLineTasksChanged = (probe, set) => _ = _hub.PushToAsync(probe, set).AsTask();
|
||
_tasks.OnPeersChanged = _hub.CloseNotIn;
|
||
_hub.OnProbeLost = probe => _tests.ProbeLost(probe);
|
||
_admission.OnEndpointChanged = probe => _ = _uplink.EmitEventAsync(AgentEventKinds.EndpointChanged, probe.UdpEndpoint?.ToString()).AsTask();
|
||
|
||
_peers = new PeerListener(
|
||
new PeerListenerOptions
|
||
{
|
||
Listen = new IPEndPoint(bind, config.Listen.ControlPort),
|
||
Certificate = identity.Certificate,
|
||
LookupPeerKey = _tasks.LookupPeerKey,
|
||
Knock = _knock,
|
||
},
|
||
_hub.OnSessionAsync,
|
||
log,
|
||
time);
|
||
|
||
_link = BuildLink();
|
||
_uplink.Attach(_link);
|
||
}
|
||
|
||
/// <summary>数据面和控制端口都开始监听之后完成（配置里的端口为 0 时，这里能读到实际端口）。</summary>
|
||
public Task Ready => _ready.Task;
|
||
|
||
public void Dispose()
|
||
{
|
||
_udp.Dispose();
|
||
_ownedRestart?.Dispose();
|
||
}
|
||
|
||
internal StreamRegistry Streams => _streams;
|
||
|
||
public IReadOnlyList<int> UdpPorts => _udp.BoundPorts;
|
||
|
||
public IReadOnlyList<int> TcpPorts => _tcp.BoundPorts;
|
||
|
||
public int ControlPort => _peers.Port;
|
||
|
||
/// <summary>节点当前采用的操作员集合版本（没有固定信任根时是 0）。</summary>
|
||
public int TrustVersion => _trust.Version;
|
||
|
||
public AgentCounters Counters => _counters;
|
||
|
||
public FleetLink Link => _link;
|
||
|
||
/// <summary>对等面（Probe 连进来的会话）的路由器，测试用。</summary>
|
||
public RpcRouter<PeerContext> PeerRouter => _hub.Router;
|
||
|
||
/// <summary>期望状态里是否已经有这个 Probe 的公钥（测试用：握手会按它校验）。</summary>
|
||
public bool IsPeerKnown(string probeId, string publicKeyBase64) =>
|
||
_tasks.LookupPeerKey(probeId) is { } key && Convert.ToBase64String(key) == publicKeyBase64;
|
||
|
||
public TestRunner Tests => _tests;
|
||
|
||
public async Task RunAsync(CancellationToken ct)
|
||
{
|
||
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||
// 数据面和控制端口先开始监听，再连接 Backend：hello 里上报的是实际端口
|
||
_udp.Bind();
|
||
_tcp.Bind();
|
||
Task udp = _udp.RunAsync(linked.Token);
|
||
Task tcp = _tcp.RunAsync(linked.Token);
|
||
Task peers = _peers.RunAsync(linked.Token);
|
||
await Task.WhenAny(_peers.Started.Task, peers);
|
||
if (peers.IsFaulted)
|
||
{
|
||
await linked.CancelAsync();
|
||
await Task.WhenAll(udp, tcp, peers); // 抛出监听失败的原因（端口被占用等）
|
||
}
|
||
|
||
_ready.TrySetResult();
|
||
_log.Info("agent listening", ("control", ControlPort), ("udp", string.Join(',', UdpPorts)), ("tcp", string.Join(',', TcpPorts)));
|
||
try
|
||
{
|
||
await Task.WhenAll(udp, tcp, peers, _link.RunAsync(linked.Token), ReportHostAsync(linked.Token));
|
||
}
|
||
finally
|
||
{
|
||
await _tasks.StopAsync();
|
||
await linked.CancelAsync();
|
||
}
|
||
}
|
||
|
||
/// <summary>每隔几秒把主机资源占用的快照发给 Backend（尽力而为：链路断开时丢弃，重连之后下一次就补上）。</summary>
|
||
private async Task ReportHostAsync(CancellationToken ct)
|
||
{
|
||
if (!_config.Reporting.HostStats)
|
||
{
|
||
return;
|
||
}
|
||
|
||
var sampler = new HostSampler(_time);
|
||
sampler.Sample(); // 第一次只用来建立 CPU 和网速的基线
|
||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(Math.Clamp(_config.Reporting.HostIntervalSeconds, 1, 60)), _time);
|
||
try
|
||
{
|
||
while (await timer.WaitForNextTickAsync(ct))
|
||
{
|
||
await _uplink.SendHostAsync(sampler.Sample());
|
||
}
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// 停止
|
||
}
|
||
}
|
||
|
||
private BudgetGuard CreateGuard() =>
|
||
new(
|
||
new NodeLimits
|
||
{
|
||
MaxTotalMbps = _config.Limits.MaxTotalMbps,
|
||
MaxTotalPps = _config.Limits.MaxTotalPps,
|
||
AllowedTargetCidrs = _config.Limits.AllowedTargetCidrs,
|
||
MaxConcurrentTests = _config.Limits.MaxConcurrentTests,
|
||
AllowedUdpPorts = _udp.BoundPorts,
|
||
AllowedTcpPorts = _tcp.BoundPorts,
|
||
},
|
||
BlockCatalog.AgentKinds);
|
||
|
||
private FleetLink BuildLink()
|
||
{
|
||
var router = new RpcRouter<LinkContext>(RpcSide.Node, MoShiroMethodTable.Table, NodeRoles.Agent);
|
||
HashSet<string>? allowed = _config.Policy.AllowMethods is { } list ? [.. list] : null;
|
||
bool Allow(string method) => allowed is null || allowed.Contains(method);
|
||
|
||
if (Allow(MoShiroMethods.TestRun))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.TestRun,
|
||
MoShiroJsonContext.Default.TestRunParams,
|
||
MoShiroJsonContext.Default.TestRunResult,
|
||
(_, run, _) => ValueTask.FromResult(_tests.Start(run)));
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.TestCancel))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.TestCancel,
|
||
MoShiroJsonContext.Default.TestCancelParams,
|
||
MoShiroJsonContext.Default.TestRunResult,
|
||
(_, cancel, _) => ValueTask.FromResult(Result<TestRunResult>.Ok(new TestRunResult { Accepted = _tests.Cancel(cancel.TestId) })));
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.UpdateOffer))
|
||
{
|
||
router.OnNotify(MoShiroMethods.UpdateOffer, MoShiroJsonContext.Default.UpdateOfferParams, (_, offer, _) =>
|
||
{
|
||
_update.WriteOffer(offer);
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.Schedule))
|
||
{
|
||
router.OnNotify(MoShiroMethods.Schedule, MoShiroJsonContext.Default.ScheduleParams, (_, schedule, _) =>
|
||
{
|
||
if (!_restart.SetRemote(schedule.RestartAt))
|
||
{
|
||
_log.Warn("ignoring an invalid restart schedule from the panel");
|
||
}
|
||
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.UpdateApply))
|
||
{
|
||
router.OnNotify(MoShiroMethods.UpdateApply, MoShiroJsonContext.Default.UpdateOfferParams, (_, _, _) =>
|
||
{
|
||
_update.RequestApply();
|
||
return ValueTask.CompletedTask;
|
||
});
|
||
}
|
||
|
||
if (Allow(MoShiroMethods.ProbeDisconnect))
|
||
{
|
||
router.On(
|
||
MoShiroMethods.ProbeDisconnect,
|
||
MoShiroJsonContext.Default.ProbeDisconnectParams,
|
||
MoShiroJsonContext.Default.TestRunResult,
|
||
async (_, request, _) => Result<TestRunResult>.Ok(new TestRunResult { Accepted = await _hub.DisconnectAsync(request.ProbeId.Value) }));
|
||
}
|
||
|
||
var options = new FleetLinkOptions
|
||
{
|
||
Server = new Uri(_config.Fleet.Server),
|
||
Identity = _identity.Identity,
|
||
ServerVerifier = _trust,
|
||
NodeId = _identity.NodeId,
|
||
NodeVersion = Version,
|
||
Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', BlockCatalog.AgentKinds)],
|
||
HelloMeta = HelloMeta,
|
||
HeartbeatExtra = HeartbeatExtra,
|
||
};
|
||
var link = new FleetLink(options, router, [new StateHandler<TaskSet>(TaskSet.StateType, MoShiroJsonContext.Default.TaskSet, _tasks.ApplyAsync)], _outbox, _time, _log);
|
||
|
||
// 链路重新连上之后补报：在线的 Probe 都告诉 Backend（Backend 重启或链路抖动期间错过了 probe_connected）
|
||
link.OnConnected = async (_, _) =>
|
||
{
|
||
foreach (ConnectedProbe probe in _hub.ConnectedProbes())
|
||
{
|
||
await _uplink.EmitEventAsync(AgentEventKinds.ProbeConnected, probe.Id.Value, probe);
|
||
}
|
||
};
|
||
return link;
|
||
}
|
||
|
||
/// <summary>敲门关卡的名册：期望状态里的 peers（任务管理器在构造之后才有）。</summary>
|
||
private IReadOnlyList<byte[]> RosterKeys() => _tasks.PeerKeys();
|
||
|
||
private JsonElement? HelloMeta()
|
||
{
|
||
var meta = new AgentHelloMeta
|
||
{
|
||
Owners = [.. _trust.Controllers],
|
||
Ipv6 = Ipv6Diagnostics.Sample(_ipv6Supported, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")),
|
||
Ipv6Supported = _ipv6Supported,
|
||
Addresses = [.. LocalAddresses.Sample()],
|
||
Platform = MoShiro.Measurement.PlatformDetector.Detect(),
|
||
TlsPublicKey = Convert.ToBase64String(_identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo()),
|
||
ProbeTcpPort = ControlPort,
|
||
ProbeUdpPorts = [.. UdpPorts],
|
||
LocalLimits = new LocalLimits
|
||
{
|
||
MaxTotalMbps = _config.Limits.MaxTotalMbps,
|
||
MaxTotalPps = _config.Limits.MaxTotalPps,
|
||
AllowedTargetCidrs = _config.Limits.AllowedTargetCidrs,
|
||
MaxConcurrentTests = _config.Limits.MaxConcurrentTests,
|
||
},
|
||
};
|
||
return JsonSerializer.SerializeToElement(meta, MoShiroJsonContext.Default.AgentHelloMeta);
|
||
}
|
||
|
||
private long _lastIn;
|
||
private long _lastOut;
|
||
private long _lastAtMs;
|
||
|
||
private JsonElement? HeartbeatExtra()
|
||
{
|
||
long nowMs = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||
long seconds = Math.Max(1, (nowMs - _lastAtMs) / 1000);
|
||
long packetsIn = _counters.PacketsIn;
|
||
long packetsOut = _counters.PacketsOut;
|
||
var extra = new AgentHeartbeatExtra
|
||
{
|
||
Ipv6 = Ipv6Diagnostics.Sample(_ipv6Supported, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")),
|
||
Probes = [.. _hub.ConnectedProbes()],
|
||
Sessions = _sessions.All.Count,
|
||
PpsIn = _lastAtMs == 0 ? 0 : (packetsIn - _lastIn) / seconds,
|
||
PpsOut = _lastAtMs == 0 ? 0 : (packetsOut - _lastOut) / seconds,
|
||
DroppedInvalid = _counters.DroppedInvalid,
|
||
DroppedRateLimited = _counters.DroppedRateLimited,
|
||
QueueBytes = _outbox.PendingBytes,
|
||
Update = _update.ReadStatus(),
|
||
};
|
||
_lastIn = packetsIn;
|
||
_lastOut = packetsOut;
|
||
_lastAtMs = nowMs;
|
||
return JsonSerializer.SerializeToElement(extra, MoShiroJsonContext.Default.AgentHeartbeatExtra);
|
||
}
|
||
}
|