justme
a2f88a344b
Join codes need no passkey touch and carry only public values (genesis fingerprint, owner credentials, installer hashes). Nodes pin the genesis and owners at enrollment and only follow passkey-signed operator set versions. The panel's server key is listed in the set; pinned nodes require its signature and run the control link under AES-256-GCM. Password sessions carry no capabilities. The updater adopts a next_key from a signed manifest. Bump version to 0.2.0. [CI SKIP]
290 lines
16 KiB
C#
290 lines
16 KiB
C#
using Fleet.Core;
|
||
using Fleet.Server;
|
||
using Microsoft.Data.Sqlite;
|
||
using MoShiro.Backend.Application;
|
||
using MoShiro.Protocol;
|
||
|
||
namespace MoShiro.Backend.Storage.Sqlite;
|
||
|
||
/// <summary>
|
||
/// Agent 表同时是 Fleet 的节点表：实现 <see cref="IAgentStore"/>（MoShiro 的视图）和 <see cref="IFleetStore"/>（注册与认证）。
|
||
/// 加入令牌的校验也在这里，和 MoShiro 的其他数据放在同一个 moshiro.db 里，由同一个单写入者写入。
|
||
/// </summary>
|
||
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, ipv6_diagnostics, public_host_manual, ipv6_host_manual, reported_addresses, tls_pubkey, added_by";
|
||
|
||
private readonly SqliteDatabase _db;
|
||
|
||
public SqliteAgentStore(SqliteDatabase db)
|
||
{
|
||
_db = db;
|
||
}
|
||
|
||
// ───────────────────────── IAgentStore ─────────────────────────
|
||
|
||
public ValueTask<IReadOnlyList<AgentRecord>> ListAsync(CancellationToken ct) =>
|
||
_db.ReadAsync<IReadOnlyList<AgentRecord>>(c => c.Query($"SELECT {AgentColumns} FROM agents ORDER BY created_at, id", ReadAgent), ct);
|
||
|
||
public ValueTask<AgentRecord?> GetAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ReadAsync(c => GetInternal(c, id), ct);
|
||
|
||
public ValueTask ApplyEnrollmentAsync(NodeId id, AgentEnrollment enrollment, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c =>
|
||
{
|
||
c.Exec(
|
||
"UPDATE agents SET public_host = CASE WHEN public_host = '' THEN @host ELSE public_host END, probe_tcp_port = @tcp, probe_udp_ports = @udp, " +
|
||
"kinds = CASE WHEN kinds = '[]' THEN @kinds ELSE kinds END, row_version = row_version + 1 WHERE id = @id",
|
||
("@id", id.Value), ("@host", enrollment.PublicHost), ("@tcp", enrollment.ProbeTcpPort),
|
||
("@udp", StorageJson.ToJson(enrollment.ProbeUdpPorts)), ("@kinds", StorageJson.ToJson(enrollment.Kinds)));
|
||
}, ct);
|
||
|
||
public ValueTask<Result<AgentRecord>> PatchAsync(NodeId id, AgentPatch patch, long? expectedRowVersion, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
AgentRecord? current = GetInternal(c, id);
|
||
if (current is null)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "agent not found");
|
||
}
|
||
|
||
if (expectedRowVersion is { } expected && expected != current.RowVersion)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.PreconditionFailed, "the agent was modified by someone else");
|
||
}
|
||
|
||
if (patch.Name is { } name && name != current.Name && current.Status != NodeStatus.Pending && NameTaken(c, name, id))
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.NameConflict, "another agent already uses this name");
|
||
}
|
||
|
||
// 手填就标记为 manual（自动识别不再覆盖）；撤销手填回到自动，值留到下一次识别
|
||
bool hostAuto = patch.PublicHostAuto == true;
|
||
bool v6Auto = patch.Ipv6HostAuto == true;
|
||
c.Exec(
|
||
"UPDATE agents SET name = @name, labels = @labels, public_host = @host, ipv6_host = @v6, public_host_manual = @hostManual, ipv6_host_manual = @v6Manual, " +
|
||
"public_visible = @visible, row_version = row_version + 1 WHERE id = @id",
|
||
("@id", id.Value), ("@name", patch.Name ?? current.Name),
|
||
("@labels", StorageJson.ToJson(patch.Labels ?? current.Labels)),
|
||
("@host", hostAuto ? current.PublicHost : patch.PublicHost ?? current.PublicHost), ("@v6", v6Auto ? current.Ipv6Host : patch.Ipv6Host ?? current.Ipv6Host),
|
||
("@hostManual", hostAuto ? 0 : patch.PublicHost is not null || current.PublicHostManual ? 1 : 0),
|
||
("@v6Manual", v6Auto ? 0 : patch.Ipv6Host is not null || current.Ipv6HostManual ? 1 : 0),
|
||
("@visible", (patch.PublicVisible ?? current.PublicVisible) ? 1 : 0));
|
||
return Result<AgentRecord>.Ok(GetInternal(c, id)!);
|
||
}, ct);
|
||
|
||
public ValueTask<Result<AgentRecord>> SetStatusAsync(NodeId id, NodeStatus status, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
AgentRecord? current = GetInternal(c, id);
|
||
if (current is null)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "agent not found");
|
||
}
|
||
|
||
if (status != NodeStatus.Pending && NameTaken(c, current.Name, id))
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.NameConflict, "another agent already uses this name; rename it or take over the existing node");
|
||
}
|
||
|
||
c.Exec("UPDATE agents SET status = @status, row_version = row_version + 1 WHERE id = @id", ("@id", id.Value), ("@status", StatusText(status)));
|
||
return Result<AgentRecord>.Ok(GetInternal(c, id)!);
|
||
}, ct);
|
||
|
||
public ValueTask<Result<AgentRecord>> TakeoverAsync(NodeId pendingId, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
AgentRecord? pending = GetInternal(c, pendingId);
|
||
if (pending is null || pending.Status != NodeStatus.Pending)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "no pending agent with this id");
|
||
}
|
||
|
||
string? existingId = c.Scalar<string>("SELECT id FROM agents WHERE name = @name AND status <> 'pending' AND id <> @id", ("@name", pending.Name), ("@id", pendingId.Value));
|
||
if (existingId is null)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "there is no established agent with this name to take over");
|
||
}
|
||
|
||
// 旧记录换成新公钥，历史数据保留；待审批记录删除
|
||
c.Exec("DELETE FROM agents WHERE id = @id", ("@id", pendingId.Value));
|
||
c.Exec(
|
||
"UPDATE agents SET pubkey = @pubkey, tls_spki = @spki, status = 'active', task_hash = NULL, row_version = row_version + 1 WHERE id = @id",
|
||
("@id", existingId), ("@pubkey", pending.PublicKey), ("@spki", pending.TlsSpki));
|
||
return Result<AgentRecord>.Ok(GetInternal(c, new NodeId(existingId))!);
|
||
}, ct);
|
||
|
||
public ValueTask DeleteAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("DELETE FROM agents WHERE id = @id", ("@id", id.Value)), ct);
|
||
|
||
public ValueTask ApplyHelloAsync(NodeId id, AgentHello hello, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c =>
|
||
{
|
||
c.Exec(
|
||
"UPDATE agents SET kinds = @kinds, agent_version = @version, local_limits = COALESCE(@limits, local_limits), tls_spki = COALESCE(@spki, tls_spki), tls_pubkey = COALESCE(@pub, tls_pubkey), " +
|
||
"probe_tcp_port = COALESCE(@tcp, probe_tcp_port), probe_udp_ports = COALESCE(@udp, probe_udp_ports), ipv6_supported = @v6, ipv6_diagnostics = @diag, " +
|
||
"reported_addresses = COALESCE(@addr, reported_addresses), 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), ("@pub", hello.TlsPublicKey), ("@tcp", hello.ProbeTcpPort), ("@udp", hello.ProbeUdpPorts is null ? null : StorageJson.ToJson(hello.ProbeUdpPorts)), ("@v6", hello.Ipv6Supported ? 1 : 0), ("@diag", hello.Ipv6DiagnosticsJson),
|
||
("@addr", hello.Addresses is null ? null : StorageJson.ToJson(hello.Addresses)));
|
||
}, ct);
|
||
|
||
public ValueTask ApplyAutoAddressesAsync(NodeId id, string? publicHost, string? ipv6Host, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c =>
|
||
{
|
||
c.Exec(
|
||
"UPDATE agents SET public_host = CASE WHEN public_host_manual = 0 AND @host IS NOT NULL THEN @host ELSE public_host END, " +
|
||
"ipv6_host = CASE WHEN ipv6_host_manual = 0 AND @v6 IS NOT NULL THEN @v6 ELSE ipv6_host END, row_version = row_version + 1 " +
|
||
"WHERE id = @id AND ((public_host_manual = 0 AND @host IS NOT NULL AND public_host <> @host) OR (ipv6_host_manual = 0 AND @v6 IS NOT NULL AND ipv6_host <> @v6))",
|
||
("@id", id.Value), ("@host", publicHost), ("@v6", ipv6Host));
|
||
}, ct);
|
||
|
||
public ValueTask SetTaskVersionAsync(NodeId id, long version, string taskHash, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("UPDATE agents SET task_version = @v, task_hash = @h WHERE id = @id", ("@id", id.Value), ("@v", version), ("@h", taskHash)), ct);
|
||
|
||
public ValueTask<string?> GetTaskHashAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ReadAsync(c => c.Scalar<string>("SELECT task_hash FROM agents WHERE id = @id", ("@id", id.Value)), ct);
|
||
|
||
public ValueTask SetAppliedTaskStateAsync(NodeId id, long appliedVersion, string? rejectedJson, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("UPDATE agents SET applied_task_version = @v, rejected = @r WHERE id = @id", ("@id", id.Value), ("@v", appliedVersion), ("@r", rejectedJson)), ct);
|
||
|
||
// ───────────────────────── IFleetStore ─────────────────────────
|
||
|
||
public ValueTask<JoinTokenRecord?> FindJoinTokenAsync(ReadOnlyMemory<byte> tokenSha256, CancellationToken ct)
|
||
{
|
||
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, kind, created_by 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), r.NullableString(11)),
|
||
("@h", hash)),
|
||
ct);
|
||
}
|
||
|
||
public ValueTask<bool> TryConsumeJoinTokenAsync(string tokenId, long nowMs, CancellationToken ct) =>
|
||
_db.WriteAsync(
|
||
c => c.Exec(
|
||
"UPDATE join_tokens SET uses = uses + 1 WHERE id = @id AND revoked_at IS NULL AND expires_at > @now AND uses < max_uses",
|
||
("@id", tokenId), ("@now", nowMs)) == 1,
|
||
ct);
|
||
|
||
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");
|
||
}
|
||
|
||
c.Exec(
|
||
"INSERT INTO agents (id, name, status, pubkey, tls_spki, labels, meta, agent_version, created_at, added_by) VALUES (@id, @name, @status, @pubkey, @spki, @labels, @meta, @version, @now, @by)",
|
||
("@id", node.Id.Value), ("@name", node.Name), ("@status", StatusText(node.Status)), ("@pubkey", node.PublicKey), ("@spki", node.TlsSpki),
|
||
("@labels", StorageJson.ToJson(node.Labels)), ("@meta", node.MetaJson), ("@version", node.NodeVersion), ("@now", node.NowMs), ("@by", node.AddedBy));
|
||
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, added_by) VALUES (@id, @name, @pubkey, @peer, @now, @by)",
|
||
("@id", node.Id.Value), ("@name", node.Name), ("@pubkey", node.PublicKey), ("@peer", peerKey), ("@now", node.NowMs), ("@by", node.AddedBy));
|
||
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
|
||
? ToNode(agent, null)
|
||
: await _db.ReadAsync(
|
||
c => c.QueryOne(
|
||
"SELECT id, name, pubkey, created_at, revoked_at FROM probes WHERE id = @id",
|
||
r => new NodeRecord(
|
||
new NodeId(r.GetString(0)), r.GetString(1), r.IsDBNull(4) ? NodeStatus.Active : NodeStatus.Disabled, r.Bytes(2), [], [], null, null, r.GetInt64(3), null),
|
||
("@id", id.Value)),
|
||
ct);
|
||
|
||
public ValueTask<NodeRecord?> FindEstablishedNodeByNameAsync(string name, CancellationToken ct) =>
|
||
_db.ReadAsync(
|
||
c => c.QueryOne($"SELECT {AgentColumns} FROM agents WHERE name = @name AND status <> 'pending'", ReadAgent, ("@name", name)) is { } agent ? ToNode(agent, null) : null,
|
||
ct);
|
||
|
||
public ValueTask UpdatePresenceAsync(NodeId id, Presence presence, CancellationToken ct) =>
|
||
_db.ExecuteAsync(
|
||
c => c.Exec(
|
||
"UPDATE agents SET last_seen_at = @at, agent_version = COALESCE(@version, agent_version) WHERE id = @id",
|
||
("@id", id.Value), ("@at", presence.AtMs), ("@version", presence.NodeVersion)),
|
||
ct);
|
||
|
||
// ───────────────────────── 内部 ─────────────────────────
|
||
|
||
private static AgentRecord? GetInternal(SqliteConnection c, NodeId id) =>
|
||
c.QueryOne($"SELECT {AgentColumns} FROM agents WHERE id = @id", ReadAgent, ("@id", id.Value));
|
||
|
||
private static bool NameTaken(SqliteConnection c, string name, NodeId except) =>
|
||
c.Scalar<long>("SELECT COUNT(*) FROM agents WHERE name = @name AND status <> 'pending' AND id <> @id", ("@name", name), ("@id", except.Value)) > 0;
|
||
|
||
private static AgentRecord ReadAgent(SqliteDataReader r) => new(
|
||
new NodeId(r.GetString(0)),
|
||
r.GetString(1),
|
||
ParseStatus(r.GetString(2)),
|
||
r.Bytes(3),
|
||
r.Bytes(4),
|
||
r.GetString(5),
|
||
r.GetInt32(6),
|
||
StorageJson.Ints(r.GetString(7)),
|
||
StorageJson.Strings(r.GetString(8)),
|
||
r.GetInt64(9) != 0,
|
||
StorageJson.Strings(r.GetString(10)),
|
||
r.NullableString(11),
|
||
r.NullableString(12),
|
||
r.GetInt64(13),
|
||
r.GetInt64(14),
|
||
r.NullableString(15),
|
||
r.GetInt64(16),
|
||
r.NullableLong(17),
|
||
r.GetInt64(18),
|
||
r.GetString(19),
|
||
r.GetInt64(20) != 0,
|
||
r.NullableString(21),
|
||
r.GetInt64(22) != 0,
|
||
r.GetInt64(23) != 0,
|
||
StorageJson.Strings(r.GetString(24)),
|
||
r.NullableBytes(25),
|
||
r.NullableString(26));
|
||
|
||
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);
|
||
|
||
private static string StatusText(NodeStatus status) => status switch
|
||
{
|
||
NodeStatus.Active => "active",
|
||
NodeStatus.Disabled => "disabled",
|
||
_ => "pending",
|
||
};
|
||
|
||
private static NodeStatus ParseStatus(string text) => text switch
|
||
{
|
||
"active" => NodeStatus.Active,
|
||
"disabled" => NodeStatus.Disabled,
|
||
_ => NodeStatus.Pending,
|
||
};
|
||
}
|