feat!: probes generate their own keys and enroll with a join token; node and peer keys are separate [CI]
ci/woodpecker/push/main-amd64-build Pipeline was successful

This commit is contained in:
justme committed 2026-10-07 16:57:34 +08:00
1 parent 1ff847e45a
commit a9f562441f
51 files changed
+505 -749

No files matched your search

@@ -40,7 +40,7 @@ public static class AgentEndpoints
group.MapGet("/join-tokens", ListTokensAsync)
.Api("join_tokens_list", "Join tokens", TokensTag).Returns<PageDto<JoinTokenDto>>().ProducesProblems(401, 403);
group.MapPost("/join-tokens", CreateTokenAsync)
.Api("join_tokens_create", "Create a join token; the plaintext is returned only once", TokensTag)
.Api("join_tokens_create", "Create a join token for an agent or a probe (role); the plaintext is returned only once", TokensTag)
.Returns<CreatedJoinTokenDto>(StatusCodes.Status201Created).ProducesProblems(400, 401, 403);
group.MapDelete("/join-tokens/{id}", RevokeTokenAsync)
.Api("join_tokens_revoke", "Revoke a join token; with purge=true delete the record for good", TokensTag)
@@ -48,16 +48,10 @@ public static class AgentEndpoints
group.MapGet("/probes", ListProbesAsync)
.Api("probes_list", "All probes, with the agents each one is connected to", ProbeTag).Returns<PageDto<ProbeDto>>().ProducesProblems(401, 403);
group.MapPost("/probes", CreateProbeAsync)
.Api("probes_create", "Create a probe identity; the private key is returned only once", ProbeTag)
.Returns<ProbeIdentityDto>(StatusCodes.Status201Created).ProducesProblems(400, 401, 403, 409);
group.MapGet("/probes/{id}", GetProbeAsync)
.Api("probes_get", "One probe", ProbeTag).Returns<ProbeDto>().ProducesProblems(401, 403, 404);
group.MapPatch("/probes/{id}", PatchProbeAsync)
.Api("probes_update", "Rename a probe", ProbeTag).Returns<ProbeDto>().ProducesProblems(400, 401, 403, 404, 409);
group.MapPost("/probes/{id}/rotate", RotateProbeAsync)
.Api("probes_rotate", "Give a probe a new key (same id): the old key stops working and its sessions are dropped; the new probe.json is returned only once", ProbeTag)
.Returns<ProbeIdentityDto>().ProducesProblems(401, 403, 404, 409);
group.MapPost("/probes/{id}/revoke", RevokeProbeAsync)
.Api("probes_revoke", "Revoke a probe: every agent drops its session and refuses it; history stays until it expires", ProbeTag)
.Returns<ProbeDto>().ProducesProblems(401, 403, 404);
@@ -122,7 +116,7 @@ public static class AgentEndpoints
{
string baseUrl = options.PublicUrl ?? $"{http.Request.Scheme}://{http.Request.Host}";
Result<CreatedJoinToken> result = await tokens.CreateAsync(
new Application.CreateJoinTokenRequest(request.Note, request.Labels, request.AutoApprove ?? true, request.MaxUses ?? 1, request.ExpiresInHours ?? 24), baseUrl, http.RequestAborted);
new Application.CreateJoinTokenRequest(request.Note, request.Labels, request.AutoApprove ?? true, request.MaxUses ?? 1, request.ExpiresInHours ?? 24, request.Role ?? JoinTokenService.AgentRole), baseUrl, http.RequestAborted);
if (!result.TryGet(out CreatedJoinToken? created, out FleetError? error))
{
return Problems.From(http, error);
@@ -150,19 +144,6 @@ public static class AgentEndpoints
private static async Task<IResult> ListProbesAsync(ProbeService probes, CancellationToken ct) =>
TypedResults.Json(new PageDto<ProbeDto>([.. (await probes.ListAsync(ct)).Select(Mappers.ToDto)], null), ApiJsonContext.Default.PageDtoProbeDto);
private static async Task<IResult> CreateProbeAsync(HttpContext http, ProbeService probes, BackendOptions options, CreateProbeRequest? request, CancellationToken ct)
{
Result<ProbeIdentityFile> result = await probes.CreateAsync(request?.Name, PanelUrl(http, options), ct);
if (!result.TryGet(out ProbeIdentityFile? file, out FleetError? error))
{
return Problems.From(http, error);
}
http.Response.Headers.CacheControl = "no-store";
http.Response.Headers.Location = "/api/admin/probes/" + file.ProbeId.Value;
return TypedResults.Json(ToIdentityDto(file), ApiJsonContext.Default.ProbeIdentityDto, statusCode: StatusCodes.Status201Created);
}
private static async Task<IResult> GetProbeAsync(HttpContext http, ProbeService probes, string id, CancellationToken ct) =>
await probes.GetAsync(id, ct) is { } view
? TypedResults.Json(Mappers.ToDto(view), ApiJsonContext.Default.ProbeDto)
@@ -179,18 +160,6 @@ public static class AgentEndpoints
return result.TryGet(out ProbeView? view, out FleetError? error) ? TypedResults.Json(Mappers.ToDto(view), ApiJsonContext.Default.ProbeDto) : Problems.From(http, error);
}
private static async Task<IResult> RotateProbeAsync(HttpContext http, ProbeService probes, BackendOptions options, string id, CancellationToken ct)
{
Result<ProbeIdentityFile> result = await probes.RotateAsync(id, PanelUrl(http, options), ct);
if (!result.TryGet(out ProbeIdentityFile? file, out FleetError? error))
{
return Problems.From(http, error);
}
http.Response.Headers.CacheControl = "no-store";
return TypedResults.Json(ToIdentityDto(file), ApiJsonContext.Default.ProbeIdentityDto);
}
private static async Task<IResult> RevokeProbeAsync(HttpContext http, ProbeService probes, string id, CancellationToken ct)
{
Result<ProbeView> result = await probes.RevokeAsync(id, ct);
@@ -213,10 +182,4 @@ public static class AgentEndpoints
return dropped ? TypedResults.NoContent() : Problems.Create(http, MoShiroErrors.ProbeOffline, "this probe has no session with this agent");
}
/// <summary>Probe 要连的面板地址:配置的公开地址,没有就用这次请求的地址。</summary>
private static string PanelUrl(HttpContext http, BackendOptions options) => (options.PublicUrl ?? $"{http.Request.Scheme}://{http.Request.Host}").TrimEnd('/');
private static ProbeIdentityDto ToIdentityDto(ProbeIdentityFile file) =>
new(file.ProbeId.Value, file.PrivateKey, file.Server, new ProbeLimitsDto(file.Limits.MaxTestMbps, file.Limits.MaxTotalPps));
}
@@ -154,7 +154,7 @@ public static class Mappers
new(t.Id.Value, t.Scope, t.AgentId?.Value, t.BaseTaskId?.Value, t.Kind, t.Target, t.Params, t.Schedule, t.Enabled, Time(t.UpdatedAtMs));
public static JoinTokenDto ToDto(JoinTokenInfo t) =>
new(t.Id, t.Note, t.Labels, t.AutoApprove, t.MaxUses, t.Uses, Time(t.ExpiresAtMs), Time(t.RevokedAtMs), Time(t.CreatedAtMs));
new(t.Id, t.Note, t.Labels, t.AutoApprove, t.MaxUses, t.Uses, Time(t.ExpiresAtMs), Time(t.RevokedAtMs), Time(t.CreatedAtMs), JoinTokenService.RoleOf(t.Kind));
public static ApiTokenDto ToDto(ApiTokenRecord t) => new(t.Id, t.Name, t.Scope, Time(t.CreatedAtMs), Time(t.LastUsedAtMs), Time(t.RevokedAtMs));
@@ -48,9 +48,10 @@ public sealed record Ipv6DiagnosticDto(bool? AddressConfigured, bool SocketSuppo
/// <summary>merge-patch:缺省或为 null 的字段不修改。</summary>
public sealed record AgentPatchRequest(string? Name, IReadOnlyList<string>? Labels, string? PublicHost, bool? PublicVisible, string? Ipv6Host = null);
public sealed record JoinTokenDto(string Id, string? Note, IReadOnlyList<string> Labels, bool AutoApprove, int MaxUses, int Uses, DateTimeOffset ExpiresAt, DateTimeOffset? RevokedAt, DateTimeOffset CreatedAt);
/// <param name="Role"><c>agent</c> 或 <c>probe</c>:用这个令牌注册出来的是哪种节点。</param>
public sealed record JoinTokenDto(string Id, string? Note, IReadOnlyList<string> Labels, bool AutoApprove, int MaxUses, int Uses, DateTimeOffset ExpiresAt, DateTimeOffset? RevokedAt, DateTimeOffset CreatedAt, string Role = "agent");
public sealed record CreateJoinTokenRequest(string? Note, IReadOnlyList<string>? Labels, bool? AutoApprove, int? MaxUses, int? ExpiresInHours);
public sealed record CreateJoinTokenRequest(string? Note, IReadOnlyList<string>? Labels, bool? AutoApprove, int? MaxUses, int? ExpiresInHours, string? Role = null);
/// <summary>令牌明文只在创建的响应里出现一次。</summary>
public sealed record InstallCommandsDto(string Linux, string Macos, string Windows);
@@ -72,13 +73,6 @@ public sealed record LinkMatrixDto(bool AutoLink, IReadOnlyList<LinkDto> Items);
public sealed record SetLinkRequest(bool? Enabled);
public sealed record ProbeLimitsDto(double MaxTestMbps, long MaxTotalPps);
/// <summary><c>probe.json</c>:含私钥,只在生成时返回一次。</summary>
public sealed record ProbeIdentityDto(string ProbeId, string PrivateKey, string Server, ProbeLimitsDto Limits);
public sealed record CreateProbeRequest(string? Name);
public sealed record PatchProbeRequest(string? Name);
public sealed record TaskDto(
@@ -56,8 +56,6 @@ namespace MoShiro.Backend.Api;
[JsonSerializable(typeof(PageDto<JoinTokenDto>))]
[JsonSerializable(typeof(CreateJoinTokenRequest))]
[JsonSerializable(typeof(CreatedJoinTokenDto))]
[JsonSerializable(typeof(ProbeIdentityDto))]
[JsonSerializable(typeof(CreateProbeRequest))]
[JsonSerializable(typeof(PatchProbeRequest))]
[JsonSerializable(typeof(ProbeDto))]
[JsonSerializable(typeof(PageDto<HostItemDto>))]
@@ -2,7 +2,8 @@ using Fleet.Core;
namespace MoShiro.Backend.Application;
public sealed record CreateJoinTokenRequest(string? Note, IReadOnlyList<string>? Labels, bool AutoApprove, int MaxUses, int ExpiresInHours);
/// <param name="Role"><c>agent</c>(默认)或 <c>probe</c>:用这个令牌注册出来的是哪种节点。</param>
public sealed record CreateJoinTokenRequest(string? Note, IReadOnlyList<string>? Labels, bool AutoApprove, int MaxUses, int ExpiresInHours, string Role = JoinTokenService.AgentRole);
/// <summary>各个系统上的安装命令(Linux 和 macOS 都用 install.sh,Windows 用 install.ps1)。</summary>
public sealed record InstallCommands(string Linux, string Macos, string Windows);
@@ -12,6 +13,14 @@ public sealed record CreatedJoinToken(JoinTokenInfo Info, string Plaintext, Inst
/// <summary>加入令牌:明文只在创建时返回一次,服务端只保存 SHA-256。</summary>
public sealed class JoinTokenService
{
public const string AgentRole = "agent";
public const string ProbeRole = "probe";
/// <summary>令牌里存的「种类」:空表示 Agent,其余就是角色名。</summary>
public static string KindOf(string role) => role == ProbeRole ? ProbeRole : "";
public static string RoleOf(string kind) => kind.Length == 0 ? AgentRole : kind;
private readonly IJoinTokenStore _store;
private readonly BackendOptions _options;
private readonly TimeProvider _time;
@@ -26,6 +35,11 @@ public sealed class JoinTokenService
public async ValueTask<Result<CreatedJoinToken>> CreateAsync(CreateJoinTokenRequest request, string? requestBaseUrl, CancellationToken ct)
{
var errors = new Dictionary<string, string>();
if (request.Role is not (AgentRole or ProbeRole))
{
errors["role"] = "must be agent or probe";
}
if (request.MaxUses is < 1 or > 1000)
{
errors["max_uses"] = "must be between 1 and 1000";
@@ -54,13 +68,18 @@ public sealed class JoinTokenService
string plaintext = TokenHash.NewToken("jt");
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
JoinTokenInfo info = await _store.CreateAsync(
Ulid.NewPrefixed("jtk", _time), TokenHash.Sha256(plaintext), request.Note, request.Labels ?? [], request.AutoApprove,
Ulid.NewPrefixed("jtk", _time), TokenHash.Sha256(plaintext), request.Note, request.Labels ?? [], request.AutoApprove, KindOf(request.Role),
request.MaxUses, now + (request.ExpiresInHours * 3_600_000L), now, ct);
string baseUrl = (_options.PublicUrl ?? requestBaseUrl ?? "https://panel.example.com").TrimEnd('/');
string shell = $"curl -fsSL {baseUrl}/install.sh | sh -s -- --join {plaintext} --name <node-name>";
// Probe 和 Agent 一样用加入令牌注册;容器节点用部署脚本,Linux 二进制用 install.sh
string shell = request.Role == ProbeRole
? $"sudo sh deploy-node.sh probe --server {baseUrl} --join {plaintext} --name <node-name>"
: $"curl -fsSL {baseUrl}/install.sh | sh -s -- --join {plaintext} --name <node-name>";
// PowerShell 6+(pwsh):在 cmd 或 PowerShell 里都能直接粘贴
string powershell = $"pwsh -NoProfile -ExecutionPolicy Bypass -Command \"& ([scriptblock]::Create((irm {baseUrl}/install.ps1))) -Join {plaintext} -Name <node-name>\"";
string powershell = request.Role == ProbeRole
? $"fleet-enroll --server {baseUrl} --join {plaintext} --name <node-name> --identity-dir <dir>"
: $"pwsh -NoProfile -ExecutionPolicy Bypass -Command \"& ([scriptblock]::Create((irm {baseUrl}/install.ps1))) -Join {plaintext} -Name <node-name>\"";
return Result<CreatedJoinToken>.Ok(new CreatedJoinToken(info, plaintext, new InstallCommands(shell, shell, powershell)));
}
@@ -20,6 +20,13 @@ public sealed class MoShiroEnrollmentHook : IEnrollmentHook
public async ValueTask OnEnrolledAsync(NodeRecord node, EnrollmentContext context, CancellationToken ct)
{
if (node.Id.Value.StartsWith(ProbeId.Prefix, StringComparison.Ordinal))
{
// Probe:新的对等面公钥要进每个 Agent 的名单,Probe 自己也要拿到 Agent 列表
await _service.AfterAgentSetChangedAsync(ct);
return;
}
AgentEnrollMeta meta = new();
if (context.ProductMeta is { } raw)
{
@@ -1,17 +1,10 @@
using System.Text.RegularExpressions;
using Fleet.Core;
using Fleet.Identity;
using Fleet.Server;
using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>本地硬上限:写在 Probe 本机的配置里,面板无法修改(在路由器上建议调低)。</summary>
public sealed record ProbeLimits(double MaxTestMbps, long MaxTotalPps);
/// <summary><c>probe.json</c>:只在生成(或轮换密钥)时返回一次,含私钥;<c>Server</c> 是 Probe 要连的面板地址。</summary>
public sealed record ProbeIdentityFile(ProbeId ProbeId, string PrivateKey, string Server, ProbeLimits Limits);
/// <summary>某个 Probe 现在连着的一个 Agent(Agent 上报)。</summary>
public sealed record ProbeConnection(NodeId AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, long SinceMs, string IpVersion = IpVersions.V4);
@@ -21,8 +14,9 @@ public sealed record ProbeView(ProbeRecord Probe, bool Online, IReadOnlyList<Pro
}
/// <summary>
/// Probe 身份(Backend 生成,私钥只返回一次):可以有多个,每个都能改名、轮换密钥、吊销、删除。
/// 所有未吊销的 Probe 的公钥都会写进每个 Agent 的期望状态(<c>peers</c>);吊销、删除、轮换密钥之后,Agent 立刻断开对应会话,Probe 到面板的连接也会被拒绝。
/// Probe:用加入令牌注册(密钥由 Probe 自己生成,Backend 从不接触私钥),可以有多个,每个都能改名、吊销、删除。
/// 所有未吊销的 Probe 的对等面公钥都会写进每个 Agent 的期望状态(<c>peers</c>);吊销、删除之后,Agent 立刻断开对应会话,Probe 到面板的连接也被断开。
/// 要换密钥就吊销后重新注册(Probe 的 id 会变)。
/// 一条线路是一对(Probe,Agent)。
/// </summary>
public sealed partial class ProbeService
@@ -46,26 +40,6 @@ public sealed partial class ProbeService
_time = time;
}
public async ValueTask<Result<ProbeIdentityFile>> CreateAsync(string? name, string server, CancellationToken ct)
{
string finalName = string.IsNullOrWhiteSpace(name) ? "probe" : name.Trim();
if (!NamePattern().IsMatch(finalName))
{
return ValidationError.Fail<ProbeIdentityFile>("name", "must be 1-64 characters: letters (any language), digits, '.', '_' or '-'");
}
if ((await _probes.ListAsync(ct)).Any(p => string.Equals(p.Name, finalName, StringComparison.OrdinalIgnoreCase)))
{
return Result<ProbeIdentityFile>.Fail(MoShiroErrors.NameConflict, $"a probe named {finalName} already exists");
}
using NodeIdentity identity = NodeIdentity.Generate();
var probe = new ProbeRecord(Ulid.NewPrefixed(ProbeId.Prefix, _time), finalName, identity.PublicKeySpki, _time.GetUtcNow().ToUnixTimeMilliseconds(), null);
await _probes.SaveAsync(probe, ct);
_gateway.NotifyAllTaskSetsChanged();
return Result<ProbeIdentityFile>.Ok(BuildFile(probe, identity, server));
}
public async ValueTask<IReadOnlyList<ProbeView>> ListAsync(CancellationToken ct)
{
IReadOnlyList<ProbeRecord> probes = await _probes.ListAsync(ct);
@@ -98,27 +72,6 @@ public sealed partial class ProbeService
return Result<ProbeView>.Ok((await GetAsync(id, ct))!);
}
/// <summary>换一把新密钥(id 不变):旧私钥立即失效,所有 Agent 上这个 Probe 的会话被断开。返回新的 probe.json(只此一次)。</summary>
public async ValueTask<Result<ProbeIdentityFile>> RotateAsync(string id, string server, CancellationToken ct)
{
if (await _probes.GetAsync(id, ct) is not { } probe)
{
return Result<ProbeIdentityFile>.Fail(MoShiroErrors.ProbeNotFound, "probe not found");
}
if (probe.RevokedAtMs is not null)
{
return Result<ProbeIdentityFile>.Fail(MoShiroErrors.ProbeRevoked, "a revoked probe cannot get a new key; create a new probe instead");
}
using NodeIdentity identity = NodeIdentity.Generate();
ProbeRecord updated = probe with { PublicKey = identity.PublicKeySpki };
await _probes.SaveAsync(updated, ct);
_gateway.NotifyAllTaskSetsChanged();
await DisconnectEverywhereAsync(id, ct);
return Result<ProbeIdentityFile>.Ok(BuildFile(updated, identity, server));
}
/// <summary>吊销:保留记录和历史数据,公钥从所有 Agent 的名单里移除,Agent 立即断开它的会话。</summary>
public async ValueTask<Result<ProbeView>> RevokeAsync(string id, CancellationToken ct)
{
@@ -164,14 +117,6 @@ public sealed partial class ProbeService
return await _gateway.DisconnectProbeAsync(agentId, new ProbeId(probeId), ct);
}
private async ValueTask DisconnectEverywhereAsync(string probeId, CancellationToken ct)
{
foreach ((NodeId agent, _) in _runtime.ConnectionsOf(probeId))
{
await _gateway.DisconnectProbeAsync(agent, new ProbeId(probeId), ct);
}
}
private ProbeView View(ProbeRecord probe, IReadOnlyList<AgentRecord> agents)
{
var connections = new List<ProbeConnection>();
@@ -187,9 +132,6 @@ public sealed partial class ProbeService
return new ProbeView(probe, _gateway.IsOnline(new NodeId(probe.Id)), [.. connections.OrderBy(c => c.AgentName, StringComparer.Ordinal)]);
}
private static ProbeIdentityFile BuildFile(ProbeRecord probe, NodeIdentity identity, string server) =>
new(new ProbeId(probe.Id), Convert.ToBase64String(identity.ExportPkcs8()), server, new ProbeLimits(MaxTestMbps: 100, MaxTotalPps: 20_000));
[GeneratedRegex(@"^[\p{L}\p{M}\p{N}._-]{1,64}$")]
private static partial Regex NamePattern();
}
@@ -64,6 +64,11 @@ public sealed record JoinTokenInfo(
int Uses,
long ExpiresAtMs,
long? RevokedAtMs,
long CreatedAtMs);
long CreatedAtMs,
string Kind = "");
public sealed record ProbeRecord(string Id, string Name, byte[] PublicKey, long CreatedAtMs, long? RevokedAtMs, string? Ipv6DiagnosticsJson = null);
/// <summary>
/// <see cref="PublicKey"/> 是连 Backend 的身份公钥,<see cref="PeerPublicKey"/> 是对等面(连 Agent)的公钥,两把是节点自己生成的独立密钥;
/// Agent 的 <c>peers</c> 里放的是后者。
/// </summary>
public sealed record ProbeRecord(string Id, string Name, byte[] PublicKey, long CreatedAtMs, long? RevokedAtMs, string? Ipv6DiagnosticsJson = null, byte[]? PeerPublicKey = null);
@@ -51,7 +51,7 @@ public interface IGeoLookup
public interface IJoinTokenStore
{
ValueTask<JoinTokenInfo> CreateAsync(string id, byte[] tokenSha256, string? note, IReadOnlyList<string> labels, bool autoApprove, int maxUses, long expiresAtMs, long nowMs, CancellationToken ct);
ValueTask<JoinTokenInfo> CreateAsync(string id, byte[] tokenSha256, string? note, IReadOnlyList<string> labels, bool autoApprove, string kind, int maxUses, long expiresAtMs, long nowMs, CancellationToken ct);
ValueTask<IReadOnlyList<JoinTokenInfo>> ListAsync(CancellationToken ct);
@@ -60,9 +60,9 @@ public sealed class TaskPlanner : IDisposable
});
}
foreach (ProbeRecord probe in probes.Where(p => p.RevokedAtMs is null && links.IsLinked(p.Id, agentId.Value)))
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.PublicKey) });
document.Peers.Add(new PeerSpec { ProbeId = new ProbeId(probe.Id), PublicKey = Convert.ToBase64String(probe.PeerPublicKey!) });
}
string hash = Convert.ToHexStringLower(SHA256.HashData(JsonSerializer.SerializeToUtf8Bytes(document, MoShiroJsonContext.Default.TaskSet)));
@@ -18,7 +18,11 @@ public static class GatewayServiceCollectionExtensions
services.AddSingleton<IAgentGateway, FleetGateway>();
services.AddHostedService<ConnectionShutdown>();
services.AddFleetEnrollment();
services.AddFleetServer(options => options.NodeIdPrefix = AgentIds.Prefix)
services.AddFleetServer(options =>
{
options.NodeIdPrefix = AgentIds.Prefix;
options.NodeIdPrefixes[JoinTokenService.ProbeRole] = ProbeId.Prefix;
})
.AddState<TaskSetStateProvider>()
.AddState<ProbeAgentsStateProvider>()
.AddInbox<RawChunkInbox>()
@@ -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 = 7;
public const int CurrentVersion = 8;
public static void Migrate(SqliteConnection connection)
{
@@ -110,6 +110,16 @@ internal static class Schema
connection.Exec("PRAGMA user_version = 7");
transaction.Commit();
}
if (version < 8)
{
// Probe 用加入令牌注册:令牌带种类;Probe 有两把独立的密钥,对等面公钥单独存
using SqliteTransaction transaction = connection.BeginTransaction();
connection.Exec("ALTER TABLE join_tokens ADD COLUMN kind TEXT NOT NULL DEFAULT ''");
connection.Exec("ALTER TABLE probes ADD COLUMN peer_pubkey BLOB");
connection.Exec("PRAGMA user_version = 8");
transaction.Commit();
}
}
/// <summary>
@@ -56,4 +56,6 @@ internal static class Sql
public static string? NullableString(this SqliteDataReader reader, int ordinal) => reader.IsDBNull(ordinal) ? null : reader.GetString(ordinal);
public static byte[] Bytes(this SqliteDataReader reader, int ordinal) => (byte[])reader.GetValue(ordinal);
public static byte[]? NullableBytes(this SqliteDataReader reader, int ordinal) => reader.IsDBNull(ordinal) ? null : (byte[])reader.GetValue(ordinal);
}
@@ -138,8 +138,8 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
byte[] hash = tokenSha256.ToArray();
return _db.ReadAsync(
c => c.QueryOne(
"SELECT id, token_sha256, note, labels, auto_approve, max_uses, uses, expires_at, revoked_at, created_at FROM join_tokens WHERE token_sha256 = @h",
r => new JoinTokenRecord(r.GetString(0), r.Bytes(1), r.NullableString(2), StorageJson.Strings(r.GetString(3)), r.GetInt64(4) != 0, r.GetInt32(5), r.GetInt32(6), r.GetInt64(7), r.NullableLong(8), r.GetInt64(9)),
"SELECT id, token_sha256, note, labels, auto_approve, max_uses, uses, expires_at, revoked_at, created_at, kind FROM join_tokens WHERE token_sha256 = @h",
r => new JoinTokenRecord(r.GetString(0), r.Bytes(1), r.NullableString(2), StorageJson.Strings(r.GetString(3)), r.GetInt64(4) != 0, r.GetInt32(5), r.GetInt32(6), r.GetInt64(7), r.NullableLong(8), r.GetInt64(9), r.GetString(10)),
("@h", hash)),
ct);
}
@@ -154,6 +154,11 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
public ValueTask<Result<NodeRecord>> CreateNodeAsync(NewNode node, CancellationToken ct) =>
_db.WriteAsync(c =>
{
if (node.Kind == JoinTokenService.ProbeRole)
{
return CreateProbe(c, node);
}
if (node.Status != NodeStatus.Pending && NameTaken(c, node.Name, node.Id))
{
return Result<NodeRecord>.Fail(FleetErrors.NameConflict, "a node with this name already exists");
@@ -166,6 +171,25 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
return Result<NodeRecord>.Ok(ToNode(GetInternal(c, node.Id)!, node.MetaJson));
}, ct);
/// <summary>Probe 用加入令牌注册:直接进 probes 表,有令牌就算批准(Probe 没有 pending 状态);两把公钥都是它自己生成的。</summary>
private static Result<NodeRecord> CreateProbe(SqliteConnection c, NewNode node)
{
if (node.TlsPublicKey is not { Length: > 0 } peerKey)
{
return Result<NodeRecord>.Fail(FleetErrors.InvalidParams, "a probe needs its peer key");
}
if (c.Scalar<long>("SELECT COUNT(*) FROM probes WHERE name = @name COLLATE NOCASE", ("@name", node.Name)) > 0)
{
return Result<NodeRecord>.Fail(FleetErrors.NameConflict, "a probe with this name already exists");
}
c.Exec(
"INSERT INTO probes (id, name, pubkey, peer_pubkey, created_at) VALUES (@id, @name, @pubkey, @peer, @now)",
("@id", node.Id.Value), ("@name", node.Name), ("@pubkey", node.PublicKey), ("@peer", peerKey), ("@now", node.NowMs));
return Result<NodeRecord>.Ok(new NodeRecord(node.Id, node.Name, NodeStatus.Active, node.PublicKey, node.TlsSpki, node.Labels, node.MetaJson, node.NodeVersion, node.NowMs, null));
}
/// <summary>Agent 在 agents 表里;Probe 的身份在 probes 表里(吊销之后按「禁用」处理,连不上)。</summary>
async ValueTask<NodeRecord?> IFleetStore.GetNodeAsync(NodeId id, CancellationToken ct) =>
await GetAsync(id, ct) is { } agent
@@ -15,21 +15,21 @@ public sealed class SqliteJoinTokenStore : IJoinTokenStore
_db = db;
}
public ValueTask<JoinTokenInfo> CreateAsync(string id, byte[] tokenSha256, string? note, IReadOnlyList<string> labels, bool autoApprove, int maxUses, long expiresAtMs, long nowMs, CancellationToken ct) =>
public ValueTask<JoinTokenInfo> CreateAsync(string id, byte[] tokenSha256, string? note, IReadOnlyList<string> labels, bool autoApprove, string kind, int maxUses, long expiresAtMs, long nowMs, CancellationToken ct) =>
_db.WriteAsync(c =>
{
c.Exec(
"INSERT INTO join_tokens (id, token_sha256, note, labels, auto_approve, max_uses, expires_at, created_at) VALUES (@id, @h, @note, @labels, @auto, @max, @exp, @now)",
("@id", id), ("@h", tokenSha256), ("@note", note), ("@labels", StorageJson.ToJson(labels)), ("@auto", autoApprove ? 1 : 0),
"INSERT INTO join_tokens (id, token_sha256, note, labels, auto_approve, kind, max_uses, expires_at, created_at) VALUES (@id, @h, @note, @labels, @auto, @kind, @max, @exp, @now)",
("@id", id), ("@h", tokenSha256), ("@note", note), ("@labels", StorageJson.ToJson(labels)), ("@auto", autoApprove ? 1 : 0), ("@kind", kind),
("@max", maxUses), ("@exp", expiresAtMs), ("@now", nowMs));
return new JoinTokenInfo(id, note, labels, autoApprove, maxUses, 0, expiresAtMs, null, nowMs);
return new JoinTokenInfo(id, note, labels, autoApprove, maxUses, 0, expiresAtMs, null, nowMs, kind);
}, ct);
public ValueTask<IReadOnlyList<JoinTokenInfo>> ListAsync(CancellationToken ct) =>
_db.ReadAsync<IReadOnlyList<JoinTokenInfo>>(
c => c.Query(
"SELECT id, note, labels, auto_approve, max_uses, uses, expires_at, revoked_at, created_at FROM join_tokens ORDER BY created_at DESC, id",
r => new JoinTokenInfo(r.GetString(0), r.NullableString(1), StorageJson.Strings(r.GetString(2)), r.GetInt64(3) != 0, r.GetInt32(4), r.GetInt32(5), r.GetInt64(6), r.NullableLong(7), r.GetInt64(8))),
"SELECT id, note, labels, auto_approve, max_uses, uses, expires_at, revoked_at, created_at, kind FROM join_tokens ORDER BY created_at DESC, id",
r => new JoinTokenInfo(r.GetString(0), r.NullableString(1), StorageJson.Strings(r.GetString(2)), r.GetInt64(3) != 0, r.GetInt32(4), r.GetInt32(5), r.GetInt64(6), r.NullableLong(7), r.GetInt64(8), r.GetString(9))),
ct);
public ValueTask<bool> RevokeAsync(string id, long nowMs, CancellationToken ct) =>
@@ -50,12 +50,12 @@ 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, 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))),
c => c.Query("SELECT id, name, pubkey, created_at, revoked_at, ipv6_diagnostics, peer_pubkey 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), r.NullableBytes(6))),
ct);
public ValueTask<ProbeRecord?> GetAsync(string id, CancellationToken ct) =>
_db.ReadAsync(
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)),
c => c.QueryOne("SELECT id, name, pubkey, created_at, revoked_at, ipv6_diagnostics, peer_pubkey 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), r.NullableBytes(6)), ("@id", id)),
ct);
public ValueTask SetIpv6DiagnosticsAsync(string id, string diagnosticsJson, CancellationToken ct) =>
@@ -71,9 +71,9 @@ public sealed class SqliteProbeStore : IProbeStore
_db.ExecuteAsync(c =>
{
c.Exec(
"INSERT INTO probes (id, name, pubkey, created_at, revoked_at) VALUES (@id, @name, @pubkey, @at, @revoked) " +
"ON CONFLICT(id) DO UPDATE SET name = excluded.name, pubkey = excluded.pubkey, created_at = excluded.created_at, revoked_at = excluded.revoked_at",
("@id", probe.Id), ("@name", probe.Name), ("@pubkey", probe.PublicKey), ("@at", probe.CreatedAtMs), ("@revoked", probe.RevokedAtMs));
"INSERT INTO probes (id, name, pubkey, peer_pubkey, created_at, revoked_at) VALUES (@id, @name, @pubkey, @peer, @at, @revoked) " +
"ON CONFLICT(id) DO UPDATE SET name = excluded.name, pubkey = excluded.pubkey, peer_pubkey = excluded.peer_pubkey, created_at = excluded.created_at, revoked_at = excluded.revoked_at",
("@id", probe.Id), ("@name", probe.Name), ("@pubkey", probe.PublicKey), ("@peer", probe.PeerPublicKey), ("@at", probe.CreatedAtMs), ("@revoked", probe.RevokedAtMs));
}, ct);
}