backend: require confirmation for signed channel updates [CI]
ci/woodpecker/pr/verify Pipeline failed
ci/woodpecker/pr/web-build unknown status

This commit is contained in:
justme committed 2026-10-07 12:57:20 +08:00
1 parent 7160de2a36
commit 84f9d211bc
62 files changed
+1717 -70

No files matched your search

@@ -1,3 +1,4 @@
using System.Text.Json;
using Fleet.Core;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
@@ -21,6 +22,12 @@ public static class BuildEndpoints
.Api("releases_list", "The latest signed release manifest of each component", Tag).Returns<PageDto<ReleaseDto>>().ProducesProblems(401, 403);
group.MapPost("/nodes/{id}/update", RequestUpdateAsync)
.Api("nodes_update", "Ask an online agent or probe to install the release it has been offered", Tag).Returns<object>(StatusCodes.Status204NoContent).ProducesProblems(401, 403, 409);
group.MapGet("/backend/update", BackendUpdateAsync)
.Api("backend_update_get", "Current channel and host-verified backend update", Tag).Returns<BackendUpdateDto>().ProducesProblems(401, 403);
group.MapPost("/backend/update", ConfirmBackendUpdateAsync)
.Api("backend_update_confirm", "Confirm one verified backend release before the host restarts it", Tag).Returns<object>(StatusCodes.Status204NoContent).ProducesProblems(400, 401, 403, 412);
group.MapGet("/backend/update/live", LiveBackendUpdateAsync)
.Api("backend_update_live", "Server-sent backend update notifications", Tag).Produces(StatusCodes.Status200OK, contentType: "text/event-stream").ProducesProblems(401, 403);
}
/// <summary>登记发布清单:挂在 <see cref="ReleaseGuard"/> 下,除了管理员,CI 用的 <c>release</c> 令牌也能调用(它只能做这一件事)。</summary>
@@ -31,7 +38,7 @@ public static class BuildEndpoints
.Returns<ReleaseDto>(StatusCodes.Status201Created).ProducesProblems(400, 401, 403, 409);
}
private static async Task<IResult> GetAsync(HttpContext http, NodeInfoService nodes, ReleaseService releases, CancellationToken ct)
private static async Task<IResult> GetAsync(HttpContext http, NodeInfoService nodes, ReleaseService releases, BackendUpdates backendUpdates, CancellationToken ct)
{
var items = (await nodes.ListAsync(ct))
.OrderBy(n => n.Kind, StringComparer.Ordinal)
@@ -39,7 +46,7 @@ public static class BuildEndpoints
.Select(n => new NodeBuildDto(n.NodeId, n.Kind, n.Os, n.OsName, n.Arch, n.Version, n.Tag, n.Commit, n.BuiltAt, ToDto(releases.StatusOf(new NodeId(n.NodeId)))))
.ToList();
http.Response.Headers.CacheControl = "no-store";
return TypedResults.Json(new BuildInfoDto(NodeEndpoints.Backend(withCommit: true), items), ApiJsonContext.Default.BuildInfoDto);
return TypedResults.Json(new BuildInfoDto(NodeEndpoints.Backend(withCommit: true), items, ToDto(await backendUpdates.GetAsync(ct))), ApiJsonContext.Default.BuildInfoDto);
}
private static async Task<IResult> ListReleasesAsync(ReleaseService releases, CancellationToken ct) =>
@@ -56,7 +63,43 @@ public static class BuildEndpoints
private static async Task<IResult> RequestUpdateAsync(HttpContext http, ReleaseService releases, string id, CancellationToken ct) =>
await releases.RequestUpdateAsync(new NodeId(id), ct) ? TypedResults.NoContent() : Problems.Create(http, MoShiroErrors.AgentOffline, "the node is not connected");
private static ReleaseDto ToDto(ReleaseOffer offer) => new(offer.Component, offer.Seq, offer.Tag, offer.Version, offer.Digest);
private static ReleaseDto ToDto(ReleaseOffer offer) => new(offer.Component, offer.Seq, offer.Tag, offer.Version, offer.Digest, offer.Channel);
private static BackendUpdateDto ToDto(BackendUpdateView view) =>
new(view.Channel, view.Enabled, view.State, view.CurrentTag, view.CurrentDigest, view.Available is { } release ? ToDto(release) : null, view.Error);
private static async Task<IResult> BackendUpdateAsync(BackendUpdates updates, CancellationToken ct) =>
TypedResults.Json(ToDto(await updates.GetAsync(ct)), ApiJsonContext.Default.BackendUpdateDto);
private static async Task<IResult> ConfirmBackendUpdateAsync(HttpContext http, BackendUpdates updates, ConfirmBackendUpdateRequest request, CancellationToken ct)
{
Result<bool> result = await updates.ConfirmAsync(request.Seq, ct);
return result.IsSuccess ? TypedResults.NoContent() : Problems.From(http, result.Error);
}
private static async Task LiveBackendUpdateAsync(HttpContext http, BackendUpdates updates, TimeProvider time, CancellationToken ct)
{
http.Response.ContentType = "text/event-stream";
http.Response.Headers.CacheControl = "no-store";
http.Response.Headers["X-Accel-Buffering"] = "no";
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(2), time);
string? previous = null;
try
{
do
{
string json = JsonSerializer.Serialize(ToDto(await updates.GetAsync(ct)), ApiJsonContext.Default.BackendUpdateDto);
await http.Response.WriteAsync(json != previous ? "event: backend_update\ndata: " + json + "\n\n" : ": heartbeat\n\n", ct);
await http.Response.Body.FlushAsync(ct);
previous = json;
}
while (await timer.WaitForNextTickAsync(ct));
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
// 浏览器断开或后端正在升级重启
}
}
private static NodeUpdateDto? ToDto(UpdateStatus? status) =>
status is null ? null : new NodeUpdateDto(status.State, status.Current, status.Available?.Tag, status.Available?.Version, status.Error, status.At > 0 ? DateTimeOffset.FromUnixTimeSeconds(status.At) : null);
@@ -41,7 +41,7 @@ public static class Mappers
public static ProbeDto ToDto(ProbeView v) =>
new(
v.Probe.Id, v.Probe.Name, Time(v.Probe.CreatedAtMs), Time(v.Probe.RevokedAtMs), v.Revoked, v.Online,
[.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs), c.IpVersion))]);
[.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs), c.IpVersion))], ProbeIpv6Of(v.Probe));
public static SeriesDto ToDto(SeriesResult r, bool live) =>
new(
@@ -103,7 +103,51 @@ public static class Mappers
: null;
return new AgentDto(
a.Id.Value, a.Name, StatusName(a.Status), view.Online, view.ProbeConnected, [.. view.Probes.Select(ToDto)], view.Host is null ? null : ToDto(view.Host), a.PublicHost, a.ProbeTcpPort, a.ProbeUdpPorts, a.Labels, a.PublicVisible,
a.Kinds, a.AgentVersion, limits, a.TaskVersion, a.AppliedTaskVersion, rejected, heartbeat, view.EndpointChanges, Time(a.CreatedAtMs), Time(a.LastSeenAtMs), a.Ipv6Host, a.Ipv6Supported);
a.Kinds, a.AgentVersion, limits, a.TaskVersion, a.AppliedTaskVersion, rejected, heartbeat, view.EndpointChanges, Time(a.CreatedAtMs), Time(a.LastSeenAtMs), a.Ipv6Host, a.Ipv6Supported, Ipv6Of(view));
}
public static Ipv6DiagnosticDto Ipv6Of(AgentView view)
{
Ipv6Diagnostic? diagnostic = view.Heartbeat?.Ipv6;
if (diagnostic is null && view.Agent.Ipv6DiagnosticsJson is { } json)
{
try
{
diagnostic = JsonSerializer.Deserialize(json, MoShiroJsonContext.Default.Ipv6Diagnostic);
}
catch (JsonException)
{
// 旧节点或损坏的快照显示未知,不推断为没有 IPv6
}
}
bool configured = view.Agent.Ipv6Host.Length > 0;
string state = diagnostic?.State ?? "unknown";
if (!configured && state is "ready" or "unknown")
{
state = "address_not_configured";
}
return new Ipv6DiagnosticDto(configured, diagnostic?.SocketSupported ?? view.Agent.Ipv6Supported, diagnostic?.Listening ?? view.Agent.Ipv6Supported,
diagnostic?.HasGlobalAddress, diagnostic?.HasDefaultRoute, state, diagnostic?.NetworkScope ?? "unknown", diagnostic?.HostHasGlobalAddress, diagnostic?.HostHasDefaultRoute);
}
private static Ipv6DiagnosticDto? ProbeIpv6Of(ProbeRecord probe)
{
if (probe.Ipv6DiagnosticsJson is not { } json)
{
return null;
}
try
{
Ipv6Diagnostic? d = JsonSerializer.Deserialize(json, MoShiroJsonContext.Default.Ipv6Diagnostic);
return d is null ? null : new Ipv6DiagnosticDto(null, d.SocketSupported, null, d.HasGlobalAddress, d.HasDefaultRoute, d.State, d.NetworkScope, d.HostHasGlobalAddress, d.HostHasDefaultRoute);
}
catch (JsonException)
{
return null;
}
}
public static TaskDto ToDto(TaskRecord t) =>
@@ -39,7 +39,11 @@ public sealed record AgentDto(
DateTimeOffset CreatedAt,
DateTimeOffset? LastSeenAt,
string Ipv6Host = "",
bool Ipv6Supported = false);
bool Ipv6Supported = false,
Ipv6DiagnosticDto? Ipv6 = null);
public sealed record Ipv6DiagnosticDto(bool? AddressConfigured, bool SocketSupported, bool? Listening, bool? HasGlobalAddress, bool? HasDefaultRoute, string State,
string NetworkScope = "unknown", bool? HostHasGlobalAddress = null, bool? HostHasDefaultRoute = null);
/// <summary>merge-patch:缺省或为 null 的字段不修改。</summary>
public sealed record AgentPatchRequest(string? Name, IReadOnlyList<string>? Labels, string? PublicHost, bool? PublicVisible, string? Ipv6Host = null);
@@ -59,7 +63,7 @@ public sealed record ConnectedProbeDto(string ProbeId, string? Version, string?
/// <summary>某个 Probe 现在连着的一个 Agent。</summary>
public sealed record ProbeConnectionDto(string AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, DateTimeOffset ConnectedSince, string IpVersion = "ipv4");
public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, bool Online, IReadOnlyList<ProbeConnectionDto> Connections);
public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, bool Online, IReadOnlyList<ProbeConnectionDto> Connections, Ipv6DiagnosticDto? Ipv6 = null);
/// <summary>一对(Probe,Agent)的线路开关:<c>linked</c> 是最终结果,<c>override</c> 是手动设置(<c>null</c> = 跟随自动连接),<c>connected</c> 是现在有没有连着。</summary>
public sealed record LinkDto(string ProbeId, string ProbeName, string AgentId, string AgentName, bool Linked, bool? Override, bool Connected);
@@ -139,8 +143,12 @@ public sealed record NodeBuildDto(string Id, string Kind, string? Os, string? Os
public sealed record NodeUpdateDto(string State, string? CurrentDigest, string? AvailableTag, string? AvailableVersion, string? Error, DateTimeOffset? At);
/// <summary>面板登记的最新发布清单(CI 签名的;验签在节点上)。</summary>
public sealed record ReleaseDto(string Component, long Seq, string Tag, string Version, string Digest);
public sealed record ReleaseDto(string Component, long Seq, string Tag, string Version, string Digest, string Channel = "release");
public sealed record PublishReleaseRequest(string? Manifest, string? Signature);
public sealed record BuildInfoDto(BuildDto Backend, IReadOnlyList<NodeBuildDto> Nodes);
public sealed record BackendUpdateDto(string Channel, bool Enabled, string State, string? CurrentTag, string? CurrentDigest, ReleaseDto? Available, string? Error);
public sealed record ConfirmBackendUpdateRequest(long Seq);
public sealed record BuildInfoDto(BuildDto Backend, IReadOnlyList<NodeBuildDto> Nodes, BackendUpdateDto? BackendUpdate = null);
@@ -35,6 +35,7 @@ namespace MoShiro.Backend.Api;
[JsonSerializable(typeof(LoginResponse))]
[JsonSerializable(typeof(MeDto))]
[JsonSerializable(typeof(AgentDto))]
[JsonSerializable(typeof(PageDto<AgentIpv6Dto>))]
[JsonSerializable(typeof(PageDto<AgentDto>))]
[JsonSerializable(typeof(AgentPatchRequest))]
[JsonSerializable(typeof(LinkDto))]
@@ -47,6 +48,8 @@ namespace MoShiro.Backend.Api;
[JsonSerializable(typeof(AboutDto))]
[JsonSerializable(typeof(NodeBuildDto))]
[JsonSerializable(typeof(BuildInfoDto))]
[JsonSerializable(typeof(BackendUpdateDto))]
[JsonSerializable(typeof(ConfirmBackendUpdateRequest))]
[JsonSerializable(typeof(NodeInfoDto))]
[JsonSerializable(typeof(PageDto<NodeInfoDto>))]
[JsonSerializable(typeof(JoinTokenDto))]
@@ -1,5 +1,8 @@
namespace MoShiro.Backend.Api;
/// <summary>IPv6 诊断只含配置与网络状态,不包含任一种 IP 地址。</summary>
public sealed record AgentIpv6Dto(string AgentId, string Name, bool Online, Ipv6DiagnosticDto Ipv6);
/// <summary>汇总:整个时间段由累加器精确合并得出的值(不是各点的平均)。</summary>
public sealed record SummaryDto(IReadOnlyDictionary<string, double?> Metrics, IReadOnlyDictionary<string, long> Samples, double Coverage);
@@ -21,6 +21,9 @@ public static class QueryEndpoints
group.MapGet("/overview", OverviewAsync)
.Api("overview_get", "Ranking of all lines over a time range", Tag)
.Returns<OverviewDto>().ProducesProblems(403, 422, 429, 503);
group.MapGet("/ipv6", Ipv6Async)
.Api("ipv6_get", "IPv6 configuration and collector diagnostics without addresses", Tag)
.Returns<PageDto<AgentIpv6Dto>>().ProducesProblems(403);
group.MapGet("/lines/{line_id}", LineAsync)
.Api("lines_get", "One line", Tag)
.Returns<LineDto>().ProducesProblems(403, 404);
@@ -41,6 +44,13 @@ public static class QueryEndpoints
.Returns<TestDto>().ProducesProblems(403, 404);
}
private static async Task<IResult> Ipv6Async(HttpContext http, AgentService agents, CancellationToken ct)
{
IReadOnlyList<AgentView> views = await agents.ListAsync(ct);
return TypedResults.Json(new PageDto<AgentIpv6Dto>([.. views.Where(v => v.Agent.Status == Fleet.Server.NodeStatus.Active && (v.Agent.PublicVisible || PublicGuard.CanSeeHidden(http)))
.Select(v => new AgentIpv6Dto(v.Agent.Id.Value, v.Agent.Name, v.Online, Mappers.Ipv6Of(v)))], null), ApiJsonContext.Default.PageDtoAgentIpv6Dto);
}
private static Task<IResult> OverviewAsync(
HttpContext http, LineService lineService, QueryService query, BackendOptions options, TimeProvider time,
string? range, DateTimeOffset? from, DateTimeOffset? to, string? preset, string? metrics, string? sort, string? order, string? ip_version) =>
@@ -53,10 +53,12 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
private readonly AgentService _service;
private readonly NodeInfoService _nodes;
private readonly ReleaseService _releases;
private readonly IProbeStore _probes;
public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases)
public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, IProbeStore probes)
{
_releases = releases;
_probes = probes;
_nodes = nodes;
_agents = agents;
_runtime = runtime;
@@ -130,6 +132,10 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
{
ProbeHelloMeta? meta = raw.Deserialize(MoShiroJsonContext.Default.ProbeHelloMeta);
await _nodes.SetPlatformAsync(id.Value, NodeKinds.Probe, meta?.Platform, ct);
if (meta?.Ipv6 is { } ipv6)
{
await _probes.SetIpv6DiagnosticsAsync(id.Value, JsonSerializer.Serialize(ipv6, MoShiroJsonContext.Default.Ipv6Diagnostic), ct);
}
}
catch (JsonException)
{
@@ -146,7 +152,12 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra:
try
{
_releases.RecordStatus(id, extra.Deserialize(MoShiroJsonContext.Default.ProbeHeartbeatExtra)?.Update);
ProbeHeartbeatExtra? heartbeat = extra.Deserialize(MoShiroJsonContext.Default.ProbeHeartbeatExtra);
_releases.RecordStatus(id, heartbeat?.Update);
if (heartbeat?.Ipv6 is { } ipv6)
{
await _probes.SetIpv6DiagnosticsAsync(id.Value, JsonSerializer.Serialize(ipv6, MoShiroJsonContext.Default.Ipv6Diagnostic), ct);
}
}
catch (JsonException)
{
@@ -196,7 +207,8 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
AgentRecord? before = await _agents.GetAsync(id, ct);
string? limitsJson = meta.LocalLimits is null ? null : JsonSerializer.Serialize(meta.LocalLimits, LocalLimitsJson.Default.LocalLimits);
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts, meta.Ipv6Supported), ct);
string? ipv6Json = meta.Ipv6 is null ? null : JsonSerializer.Serialize(meta.Ipv6, MoShiroJsonContext.Default.Ipv6Diagnostic);
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts, meta.Ipv6Supported, ipv6Json), ct);
// 端口或证书指纹变了,Probe 需要新的列表
AgentRecord? after = await _agents.GetAsync(id, ct);
@@ -128,6 +128,8 @@ public sealed partial class ProbeService
}
await _probes.RevokeAsync(id, _time.GetUtcNow().ToUnixTimeMilliseconds(), ct);
// 吊销会阻止未来认证;也断开当前直连 Backend 的长会话,不等待 Probe 心跳超时。
await _gateway.DisconnectAsync(new NodeId(id), "probe revoked", ct);
_gateway.NotifyAllTaskSetsChanged();
return Result<ProbeView>.Ok((await GetAsync(id, ct))!);
}
@@ -37,6 +37,8 @@ public static class ApplicationServiceCollectionExtensions
services.AddSingleton<AuthService>();
services.AddSingleton<SettingsService>();
services.AddSingleton<ReleaseService>();
services.AddSingleton<BackendUpdates>();
services.AddHostedService(p => p.GetRequiredService<BackendUpdates>());
services.AddSingleton<JoinTokenService>();
services.AddSingleton<RawBrowserService>();
services.AddSingleton<EventService>();
@@ -6,6 +6,18 @@ public sealed class BackendOptions
/// <summary>数据目录:<c>moshiro.db</c> 与 <c>raw/</c>。</summary>
public string DataDirectory { get; set; } = "data";
/// <summary>Backend 更新通道:auto 按镜像的构建 tag 识别 dev/release,Compose 可以显式覆盖。</summary>
public string UpdateChannel { get; set; } = "auto";
/// <summary>容器使用的 OCI repository namespace;Compose 默认 moshiro,开发镜像使用 moshiro-dev。</summary>
public string ImageNamespace { get; set; } = "";
/// <summary>容器内更新请求目录;留空使用数据目录下的 update/。</summary>
public string? BackendUpdateDirectory { get; set; }
/// <summary>宿主机更新器的只读状态目录;留空使用数据目录下的 update-status/。</summary>
public string? BackendUpdateStatusDirectory { get; set; }
/// <summary>面板的对外地址(用于生成安装命令)。</summary>
public string? PublicUrl { get; set; }
@@ -24,7 +24,8 @@ public sealed record AgentRecord(
long? LastSeenAtMs,
long RowVersion,
string Ipv6Host = "",
bool Ipv6Supported = false);
bool Ipv6Supported = false,
string? Ipv6DiagnosticsJson = null);
/// <summary>管理员可以修改的字段(merge-patch 语义:为 null 的字段不改)。</summary>
public sealed class AgentPatch
@@ -48,7 +49,8 @@ public sealed record AgentHello(
byte[]? TlsSpki,
int? ProbeTcpPort,
IReadOnlyList<int>? ProbeUdpPorts,
bool Ipv6Supported = false);
bool Ipv6Supported = false,
string? Ipv6DiagnosticsJson = null);
/// <summary>注册之后 MoShiro 的补充:公开地址和端口。</summary>
public sealed record AgentEnrollment(string PublicHost, int ProbeTcpPort, IReadOnlyList<int> ProbeUdpPorts, IReadOnlyList<string> Kinds);
@@ -64,4 +66,4 @@ public sealed record JoinTokenInfo(
long? RevokedAtMs,
long CreatedAtMs);
public sealed record ProbeRecord(string Id, string Name, byte[] PublicKey, long CreatedAtMs, long? RevokedAtMs);
public sealed record ProbeRecord(string Id, string Name, byte[] PublicKey, long CreatedAtMs, long? RevokedAtMs, string? Ipv6DiagnosticsJson = null);
@@ -69,6 +69,8 @@ public interface IProbeStore
/// <summary>新增或整条更新(轮换密钥、改名都走这里)。</summary>
ValueTask SaveAsync(ProbeRecord probe, CancellationToken ct);
ValueTask SetIpv6DiagnosticsAsync(string id, string diagnosticsJson, CancellationToken ct);
/// <summary>吊销:保留记录,公钥立即从所有 Agent 的名单里移除。已经吊销的返回 false。</summary>
ValueTask<bool> RevokeAsync(string id, long nowMs, CancellationToken ct);
@@ -0,0 +1,118 @@
using Fleet.Core;
using Microsoft.Extensions.Hosting;
using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>宿主机更新器的文件信箱端口。面板只写签名清单和确认的 seq,不执行命令。</summary>
public interface IBackendUpdateFiles
{
ValueTask WriteOfferAsync(string manifest, string signature, CancellationToken ct);
ValueTask<UpdateStatus?> ReadStatusAsync(CancellationToken ct);
ValueTask<long?> ReadRequestAsync(CancellationToken ct);
ValueTask RequestAsync(long seq, CancellationToken ct);
}
public sealed record BackendUpdateView(string Channel, bool Enabled, string State, string? CurrentTag, string? CurrentDigest, ReleaseOffer? Available, string? Error);
/// <summary>Backend 的发布通知和确认。新清单只显示为可用更新,宿主机必须收到对应 seq 的确认才会重启。</summary>
public sealed class BackendUpdates : IHostedService
{
private readonly ReleaseService _releases;
private readonly IBackendUpdateFiles _files;
private readonly BackendOptions _options;
public BackendUpdates(ReleaseService releases, IBackendUpdateFiles files, BackendOptions options)
{
_releases = releases;
_files = files;
_options = options;
}
public static string ChannelOf(BackendOptions options)
{
if (options.UpdateChannel is "dev" or "release")
{
return options.UpdateChannel;
}
if (options.UpdateChannel != "auto")
{
throw new InvalidOperationException("UpdateChannel must be auto, dev or release");
}
string imageNamespace = options.ImageNamespace.Trim('/');
if (imageNamespace == "moshiro-dev" || imageNamespace.EndsWith("/moshiro-dev", StringComparison.Ordinal))
{
return "dev";
}
if (imageNamespace == "moshiro" || imageNamespace.EndsWith("/moshiro", StringComparison.Ordinal))
{
return "release";
}
return BuildInfo.Tag is { } tag && (tag.StartsWith("dev-", StringComparison.Ordinal) || tag.StartsWith("main-", StringComparison.Ordinal)) ? "dev" : "release";
}
public async ValueTask<BackendUpdateView> GetAsync(CancellationToken ct)
{
string channel = ChannelOf(_options);
ReleaseOffer? offer = await _releases.GetBackendAsync(channel, ct);
UpdateStatus? status = await _files.ReadStatusAsync(ct);
bool installed = offer is not null && (status?.Current == offer.Digest || BuildInfo.Tag == offer.Tag);
// 只有更新器确认过同一份清单才允许按钮执行;旧状态不能被套到新的清单上。
bool verified = offer is not null && status?.Available?.Seq == offer.Seq && status.Available.Digest == offer.Digest;
string state = status?.State ?? "unavailable";
if (installed)
{
state = "up_to_date";
}
else if (offer is not null && !verified && state is not ("invalid" or "failed" or "rolled_back"))
{
state = status is null ? "unavailable" : "checking";
}
if (verified && state == "available" && await _files.ReadRequestAsync(ct) == offer!.Seq)
{
state = "requested";
}
return new BackendUpdateView(channel, status is not null, state, BuildInfo.Tag, status?.Current, installed ? null : offer, status?.Error);
}
public async ValueTask<Result<bool>> ConfirmAsync(long seq, CancellationToken ct)
{
// 与 webhook 共用发布锁:确认之后即使又来了一个 tag,信箱中的请求仍只认本次确认的 seq。
await _releases.BackendGate.WaitAsync(ct);
try
{
BackendUpdateView view = await GetAsync(ct);
if (!view.Enabled || view.State != "available" || view.Available?.Seq != seq)
{
return Result<bool>.Fail(MoShiroErrors.PreconditionFailed, "the offered release changed or has not been verified by the host updater; refresh before confirming");
}
await _files.RequestAsync(seq, ct);
return Result<bool>.Ok(true);
}
finally
{
_releases.BackendGate.Release();
}
}
public async Task StartAsync(CancellationToken cancellationToken)
{
string channel = ChannelOf(_options);
if (await _releases.GetBackendAsync(channel, cancellationToken) is { } offer)
{
await _files.WriteOfferAsync(offer.Manifest, offer.Signature, cancellationToken);
}
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}
@@ -8,29 +8,38 @@ using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>面板记下的、CI 签名的发布清单(只留每个组件最新的一份)。<c>Manifest</c> 是签名对象的原始文本,面板一个字节也不改。</summary>
public sealed record ReleaseOffer(string Component, long Seq, string Tag, string Version, string Digest, string Manifest, string Signature);
public sealed record ReleaseOffer(string Component, long Seq, string Tag, string Version, string Digest, string Manifest, string Signature, string Channel = "release");
/// <summary>
/// 发布清单的分发:CI 签名之后推给面板,面板存下来并原样转给在线的 Agent、Probe(节点连上时再补发一次)。
/// <b>面板不验签、也不决定拉什么</b>:节点宿主机上的更新器用本机的发布公钥验签,并且只认本机配置的镜像仓库;
/// 所以面板即使被攻破,也推不出一个能被装上的假镜像。面板只负责「通知」和「请求更新」,状态由节点随心跳报回来。
/// </summary>
public sealed partial class ReleaseService
public sealed partial class ReleaseService : IDisposable
{
public const int MaxManifestBytes = 4096;
public const int MaxSignatureBytes = 512;
private static readonly string[] s_components = [NodeKinds.Agent, NodeKinds.Probe];
public const string BackendComponent = "backend";
private static readonly string[] s_components = [NodeKinds.Agent, NodeKinds.Probe, BackendComponent];
private readonly ISettingsStore _settings;
private readonly IAgentGateway _gateway;
private readonly IBackendUpdateFiles _backendFiles;
private readonly BackendOptions _options;
private readonly ConcurrentDictionary<string, UpdateStatus> _status = new();
public ReleaseService(ISettingsStore settings, IAgentGateway gateway)
public ReleaseService(ISettingsStore settings, IAgentGateway gateway, IBackendUpdateFiles backendFiles, BackendOptions options)
{
_settings = settings;
_gateway = gateway;
_backendFiles = backendFiles;
_options = options;
}
internal SemaphoreSlim BackendGate { get; } = new(1, 1);
public void Dispose() => BackendGate.Dispose();
/// <summary>登记一份新清单并推给在线的同类节点。只要格式对、<c>seq</c> 比现有的新就收下(签名由节点验证)。</summary>
public async ValueTask<Result<ReleaseOffer>> PublishAsync(string manifest, string signature, CancellationToken ct)
{
@@ -39,20 +48,40 @@ public sealed partial class ReleaseService
return ValidationError.Fail<ReleaseOffer>("manifest", problem);
}
ReleaseOffer? current = await GetAsync(offer.Component, ct);
if (current is not null && offer.Seq <= current.Seq)
await BackendGate.WaitAsync(ct);
try
{
return Result<ReleaseOffer>.Fail(MoShiroErrors.NameConflict, $"a release with seq {current.Seq} or newer is already registered");
}
bool backend = offer.Component == BackendComponent;
ReleaseOffer? current = backend ? await GetBackendAsync(offer.Channel, ct) : await GetAsync(offer.Component, ct);
if (current is not null && offer.Seq <= current.Seq)
{
return Result<ReleaseOffer>.Fail(MoShiroErrors.NameConflict, $"a release with seq {current.Seq} or newer is already registered");
}
await _settings.SetAsync(Key(offer.Component, "manifest"), offer.Manifest, ct);
await _settings.SetAsync(Key(offer.Component, "signature"), offer.Signature, ct);
foreach (NodeId node in OnlineOf(offer.Component))
string componentKey = backend ? BackendComponent + "." + offer.Channel : offer.Component;
await _settings.SetAsync(Key(componentKey, "manifest"), offer.Manifest, ct);
await _settings.SetAsync(Key(componentKey, "signature"), offer.Signature, ct);
if (backend)
{
if (offer.Channel == BackendUpdates.ChannelOf(_options))
{
await _backendFiles.WriteOfferAsync(offer.Manifest, offer.Signature, ct);
}
}
else
{
foreach (NodeId node in OnlineOf(offer.Component))
{
await _gateway.NotifyUpdateOfferAsync(node, offer.Manifest, offer.Signature, ct);
}
}
return Result<ReleaseOffer>.Ok(offer);
}
finally
{
await _gateway.NotifyUpdateOfferAsync(node, offer.Manifest, offer.Signature, ct);
BackendGate.Release();
}
return Result<ReleaseOffer>.Ok(offer);
}
public async ValueTask<ReleaseOffer?> GetAsync(string component, CancellationToken ct)
@@ -67,7 +96,8 @@ public sealed partial class ReleaseService
var list = new List<ReleaseOffer>();
foreach (string component in s_components)
{
if (await GetAsync(component, ct) is { } offer)
ReleaseOffer? offer = component == BackendComponent ? await GetBackendAsync(BackendUpdates.ChannelOf(_options), ct) : await GetAsync(component, ct);
if (offer is not null)
{
list.Add(offer);
}
@@ -76,6 +106,15 @@ public sealed partial class ReleaseService
return list;
}
public async ValueTask<ReleaseOffer?> GetBackendAsync(string channel, CancellationToken ct)
{
string key = BackendComponent + "." + channel;
string? manifest = await _settings.GetAsync(Key(key, "manifest"), ct);
string? signature = await _settings.GetAsync(Key(key, "signature"), ct);
return manifest is not null && signature is not null && TryParse(manifest, signature, out ReleaseOffer? offer, out _)
&& offer.Component == BackendComponent && offer.Channel == channel ? offer : null;
}
/// <summary>节点刚连上:把它这个组件最新的清单补发一份(节点离线时发布的也不会错过)。</summary>
public async ValueTask OfferToNodeAsync(NodeId node, CancellationToken ct)
{
@@ -129,7 +168,14 @@ public sealed partial class ReleaseService
JsonElement root = document.RootElement;
string component = Text(root, "component");
string digest = Text(root, "digest");
string channel = Text(root, "channel");
if (channel.Length == 0)
{
channel = "release"; // 老节点的清单没有 channel,保留原来的兼容性
}
if (root.ValueKind != JsonValueKind.Object || Array.IndexOf(s_components, component) < 0
|| channel is not ("dev" or "release")
|| !root.TryGetProperty("seq", out JsonElement seq) || !seq.TryGetInt64(out long seqValue) || seqValue <= 0
|| !Digest().IsMatch(digest) || Text(root, "image").Length is 0 or > 200)
{
@@ -137,7 +183,7 @@ public sealed partial class ReleaseService
return false;
}
offer = new ReleaseOffer(component, seqValue, Text(root, "tag"), Text(root, "version"), digest, manifest, signature);
offer = new ReleaseOffer(component, seqValue, Text(root, "tag"), Text(root, "version"), digest, manifest, signature, channel);
return true;
}
catch (JsonException)
@@ -0,0 +1,89 @@
using System.Globalization;
using System.Text;
using System.Text.Json;
using MoShiro.Backend.Application;
using MoShiro.Protocol;
namespace MoShiro.Backend.Storage.Sqlite;
/// <summary>Backend 到宿主机的文件信箱。状态卷只读挂载,业务进程改写不了验签和安装结果。</summary>
public sealed class BackendUpdateFiles : IBackendUpdateFiles
{
private readonly string _requests;
private readonly string _status;
public BackendUpdateFiles(BackendOptions options)
{
_requests = Path.GetFullPath(options.BackendUpdateDirectory ?? Path.Combine(options.DataDirectory, "update"));
_status = Path.GetFullPath(options.BackendUpdateStatusDirectory ?? Path.Combine(options.DataDirectory, "update-status"));
}
public async ValueTask WriteOfferAsync(string manifest, string signature, CancellationToken ct)
{
await WriteAsync("offer.manifest", manifest, ct);
await WriteAsync("offer.sig", signature, ct);
}
public async ValueTask<UpdateStatus?> ReadStatusAsync(CancellationToken ct)
{
string? json = await ReadAsync(Path.Combine(_status, "status.json"), 4096, ct);
if (json is null)
{
return null;
}
try
{
return JsonSerializer.Deserialize(json, MoShiroJsonContext.Default.UpdateStatus);
}
catch (JsonException)
{
return null;
}
}
public async ValueTask<long?> ReadRequestAsync(CancellationToken ct)
{
string? request = await ReadAsync(Path.Combine(_requests, "apply.request"), 64, ct);
return long.TryParse(request, NumberStyles.None, CultureInfo.InvariantCulture, out long seq) ? seq : null;
}
public ValueTask RequestAsync(long seq, CancellationToken ct) => WriteAsync("apply.request", seq.ToString(CultureInfo.InvariantCulture), ct);
private async ValueTask WriteAsync(string name, string value, CancellationToken ct)
{
Directory.CreateDirectory(_requests);
string path = Path.Combine(_requests, name);
string temporary = Path.Combine(_requests, $".{name}.{Guid.NewGuid():N}");
try
{
await File.WriteAllTextAsync(temporary, value, new UTF8Encoding(false), ct);
File.Move(temporary, path, overwrite: true);
}
finally
{
if (File.Exists(temporary))
{
File.Delete(temporary);
}
}
}
private static async ValueTask<string?> ReadAsync(string path, int limit, CancellationToken ct)
{
try
{
var file = new FileInfo(path);
if (!file.Exists || file.LinkTarget is not null || file.Length > limit)
{
return null;
}
return await File.ReadAllTextAsync(path, ct);
}
catch (IOException)
{
return null; // 更新器正做原子替换,下一次读取再试
}
}
}
@@ -5,7 +5,7 @@ namespace MoShiro.Backend.Storage.Sqlite;
/// <summary>表结构。版本记录在 <c>PRAGMA user_version</c>;以后的迁移是按版本号追加的内置 SQL。</summary>
internal static class Schema
{
public const int CurrentVersion = 6;
public const int CurrentVersion = 7;
public static void Migrate(SqliteConnection connection)
{
@@ -101,6 +101,15 @@ internal static class Schema
connection.Exec("PRAGMA user_version = 6");
transaction.Commit();
}
if (version < 7)
{
using SqliteTransaction transaction = connection.BeginTransaction();
connection.Exec("ALTER TABLE agents ADD COLUMN ipv6_diagnostics TEXT");
connection.Exec("ALTER TABLE probes ADD COLUMN ipv6_diagnostics TEXT");
connection.Exec("PRAGMA user_version = 7");
transaction.Commit();
}
}
/// <summary>
@@ -14,7 +14,7 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
{
private const string AgentColumns =
"id, name, status, pubkey, tls_spki, public_host, probe_tcp_port, probe_udp_ports, labels, public_visible, kinds, agent_version, local_limits, " +
"task_version, applied_task_version, rejected, created_at, last_seen_at, row_version, ipv6_host, ipv6_supported";
"task_version, applied_task_version, rejected, created_at, last_seen_at, row_version, ipv6_host, ipv6_supported, ipv6_diagnostics";
private readonly SqliteDatabase _db;
@@ -117,9 +117,9 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
{
c.Exec(
"UPDATE agents SET kinds = @kinds, agent_version = @version, local_limits = COALESCE(@limits, local_limits), tls_spki = COALESCE(@spki, tls_spki), " +
"probe_tcp_port = COALESCE(@tcp, probe_tcp_port), probe_udp_ports = COALESCE(@udp, probe_udp_ports), ipv6_supported = @v6, row_version = row_version + 1 WHERE id = @id",
"probe_tcp_port = COALESCE(@tcp, probe_tcp_port), probe_udp_ports = COALESCE(@udp, probe_udp_ports), ipv6_supported = @v6, ipv6_diagnostics = @diag, row_version = row_version + 1 WHERE id = @id",
("@id", id.Value), ("@kinds", StorageJson.ToJson(hello.Kinds)), ("@version", hello.AgentVersion), ("@limits", hello.LocalLimitsJson),
("@spki", hello.TlsSpki), ("@tcp", hello.ProbeTcpPort), ("@udp", hello.ProbeUdpPorts is null ? null : StorageJson.ToJson(hello.ProbeUdpPorts)), ("@v6", hello.Ipv6Supported ? 1 : 0));
("@spki", hello.TlsSpki), ("@tcp", hello.ProbeTcpPort), ("@udp", hello.ProbeUdpPorts is null ? null : StorageJson.ToJson(hello.ProbeUdpPorts)), ("@v6", hello.Ipv6Supported ? 1 : 0), ("@diag", hello.Ipv6DiagnosticsJson));
}, ct);
public ValueTask SetTaskVersionAsync(NodeId id, long version, string taskHash, CancellationToken ct) =>
@@ -219,7 +219,8 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
r.NullableLong(17),
r.GetInt64(18),
r.GetString(19),
r.GetInt64(20) != 0);
r.GetInt64(20) != 0,
r.NullableString(21));
private static NodeRecord ToNode(AgentRecord a, string? metaJson) =>
new(a.Id, a.Name, a.Status, a.PublicKey, a.TlsSpki, a.Labels, metaJson, a.AgentVersion, a.CreatedAtMs, a.LastSeenAtMs);
@@ -50,14 +50,17 @@ public sealed class SqliteProbeStore : IProbeStore
public ValueTask<IReadOnlyList<ProbeRecord>> ListAsync(CancellationToken ct) =>
_db.ReadAsync<IReadOnlyList<ProbeRecord>>(
c => c.Query("SELECT id, name, pubkey, created_at, revoked_at FROM probes ORDER BY created_at, id", r => new ProbeRecord(r.GetString(0), r.GetString(1), r.Bytes(2), r.GetInt64(3), r.NullableLong(4))),
c => c.Query("SELECT id, name, pubkey, created_at, revoked_at, ipv6_diagnostics FROM probes ORDER BY created_at, id", r => new ProbeRecord(r.GetString(0), r.GetString(1), r.Bytes(2), r.GetInt64(3), r.NullableLong(4), r.NullableString(5))),
ct);
public ValueTask<ProbeRecord?> GetAsync(string id, CancellationToken ct) =>
_db.ReadAsync(
c => c.QueryOne("SELECT id, name, pubkey, created_at, revoked_at FROM probes WHERE id = @id", r => new ProbeRecord(r.GetString(0), r.GetString(1), r.Bytes(2), r.GetInt64(3), r.NullableLong(4)), ("@id", id)),
c => c.QueryOne("SELECT id, name, pubkey, created_at, revoked_at, ipv6_diagnostics FROM probes WHERE id = @id", r => new ProbeRecord(r.GetString(0), r.GetString(1), r.Bytes(2), r.GetInt64(3), r.NullableLong(4), r.NullableString(5)), ("@id", id)),
ct);
public ValueTask SetIpv6DiagnosticsAsync(string id, string diagnosticsJson, CancellationToken ct) =>
_db.ExecuteAsync(c => c.Exec("UPDATE probes SET ipv6_diagnostics = @diag WHERE id = @id", ("@diag", diagnosticsJson), ("@id", id)), ct);
public ValueTask<bool> RevokeAsync(string id, long nowMs, CancellationToken ct) =>
_db.WriteAsync(c => c.Exec("UPDATE probes SET revoked_at = @now WHERE id = @id AND revoked_at IS NULL", ("@id", id), ("@now", nowMs)) == 1, ct);
@@ -25,6 +25,7 @@ public static class StorageServiceCollectionExtensions
services.AddSingleton<IRawStore, SqliteRawStore>();
services.AddSingleton<IMetricStore, SqliteMetricStore>();
services.AddSingleton<IStorageHealth, SqliteStorageHealth>();
services.AddSingleton<IBackendUpdateFiles, BackendUpdateFiles>();
return services;
}
}