feat!: pre-handshake knock on the agent TCP ports; probes get the agent certificate public key [CI]
ci/woodpecker/push/main-amd64-build Pipeline was successful
ci/woodpecker/push/main-amd64-build Pipeline was successful
This commit is contained in:
1 parent
f03988d68b
commit
b99b95c738
31 files changed
+539
-61
No files matched your search
@@ -1,3 +1,4 @@
|
||||
using Fleet.Identity;
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using Fleet.Core;
|
||||
@@ -16,6 +17,7 @@ public sealed class AgentApp : IDisposable
|
||||
|
||||
private readonly AgentConfig _config;
|
||||
private readonly bool _ipv6Supported;
|
||||
private readonly KnockGate _knock;
|
||||
private readonly RestartController _restart;
|
||||
private readonly RestartController? _ownedRestart;
|
||||
private readonly LoadedNodeIdentity _identity;
|
||||
@@ -60,7 +62,9 @@ public sealed class AgentApp : IDisposable
|
||||
_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);
|
||||
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data);
|
||||
// 预握手敲门:控制端口和数据端口共用一个关卡(限速和去重一起算);名册就是期望状态里的 peers
|
||||
_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
|
||||
@@ -92,6 +96,7 @@ public sealed class AgentApp : IDisposable
|
||||
Listen = new IPEndPoint(bind, config.Listen.ControlPort),
|
||||
Certificate = identity.Certificate,
|
||||
LookupPeerKey = _tasks.LookupPeerKey,
|
||||
Knock = _knock,
|
||||
},
|
||||
_hub.OnSessionAsync,
|
||||
log,
|
||||
@@ -284,6 +289,9 @@ public sealed class AgentApp : IDisposable
|
||||
return link;
|
||||
}
|
||||
|
||||
/// <summary>敲门关卡的名册:期望状态里的 peers(任务管理器在构造之后才有)。</summary>
|
||||
private IReadOnlyList<byte[]> RosterKeys() => _tasks.PeerKeys();
|
||||
|
||||
private JsonElement? HelloMeta()
|
||||
{
|
||||
var meta = new AgentHelloMeta
|
||||
@@ -292,7 +300,7 @@ public sealed class AgentApp : IDisposable
|
||||
Ipv6Supported = _ipv6Supported,
|
||||
Addresses = [.. LocalAddresses.Sample()],
|
||||
Platform = MoShiro.Measurement.PlatformDetector.Detect(),
|
||||
TlsSpki = _identity.File.TlsSpki,
|
||||
TlsPublicKey = Convert.ToBase64String(_identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo()),
|
||||
ProbeTcpPort = ControlPort,
|
||||
ProbeUdpPorts = [.. UdpPorts],
|
||||
LocalLimits = new LocalLimits
|
||||
|
||||
@@ -52,6 +52,9 @@ public sealed class TaskManager
|
||||
/// <summary>点对点握手时查找 Probe 的公钥(SPKI)。</summary>
|
||||
public byte[]? LookupPeerKey(string peerId) => _peers.TryGetValue(peerId, out byte[]? key) ? key : null;
|
||||
|
||||
/// <summary>名册里全部 Probe 的公钥:敲门关卡用它们派生配对密钥。</summary>
|
||||
public IReadOnlyList<byte[]> PeerKeys() => [.. _peers.Values];
|
||||
|
||||
/// <summary>一个新的 Probe 会话开始:为它登记全部线路任务的流并启动监听方积木,返回要下发给它的线路任务集。</summary>
|
||||
public LineTaskSet OpenSession(ProbeSession probe)
|
||||
{
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
using System.Buffers.Binary;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
|
||||
namespace Fleet.Identity;
|
||||
|
||||
/// <summary>
|
||||
/// 预握手敲门:TCP 连接建立之后、TLS 握手之前,拨号方先发 32 字节 <c>nonce(16) | tag(16)</c>,
|
||||
/// <c>tag = HMAC-SHA256(K_pair, 时间窗 | nonce)</c> 截断到 16 字节。监听方核对不过就什么都不写、直接断开,
|
||||
/// 所以扫描器拿不到证书、TLS 参数和 JARM 指纹。<c>K_pair</c> 只有名册里的成员算得出:双方身份密钥做静态 ECDH,再经 HKDF 加用途标签派生。
|
||||
/// TCP 的 SYN-ACK 由内核发出,端口在扫描器眼里仍是「开着但不说话」,不是「没监听」。
|
||||
/// </summary>
|
||||
public static class Knock
|
||||
{
|
||||
public const int PacketBytes = 32;
|
||||
public const int NonceBytes = 16;
|
||||
public const int TagBytes = 16;
|
||||
|
||||
/// <summary>时间窗长度;接受前后各一个窗,容忍时钟偏差。</summary>
|
||||
public const int WindowSeconds = 30;
|
||||
|
||||
private const string Label = "moshiro-knock-v1";
|
||||
|
||||
/// <summary>两把身份密钥的配对密钥。info 里的两个公钥按字节序排列,所以两边算出的是同一个值。</summary>
|
||||
public static byte[] DeriveKey(NodeIdentity own, ReadOnlySpan<byte> peerSpki)
|
||||
{
|
||||
byte[] secret = own.AgreeWith(peerSpki);
|
||||
try
|
||||
{
|
||||
ReadOnlySpan<byte> mine = own.PublicKeySpki;
|
||||
bool mineFirst = mine.SequenceCompareTo(peerSpki) <= 0;
|
||||
byte[] info = [.. Encoding.UTF8.GetBytes(Label), .. (mineFirst ? mine : peerSpki), .. (mineFirst ? peerSpki : mine)];
|
||||
return HKDF.DeriveKey(HashAlgorithmName.SHA256, secret, 32, null, info);
|
||||
}
|
||||
finally
|
||||
{
|
||||
CryptographicOperations.ZeroMemory(secret);
|
||||
}
|
||||
}
|
||||
|
||||
public static byte[] Build(ReadOnlySpan<byte> key, long nowMs)
|
||||
{
|
||||
byte[] packet = new byte[PacketBytes];
|
||||
RandomNumberGenerator.Fill(packet.AsSpan(0, NonceBytes));
|
||||
Tag(key, WindowOf(nowMs), packet.AsSpan(0, NonceBytes), packet.AsSpan(NonceBytes));
|
||||
return packet;
|
||||
}
|
||||
|
||||
/// <summary>tag 在前一个、当前、后一个时间窗里有一个对得上就通过。不管重放(重放由 <see cref="KnockGate"/> 按 nonce 去重)。</summary>
|
||||
public static bool Check(ReadOnlySpan<byte> key, ReadOnlySpan<byte> packet, long nowMs)
|
||||
{
|
||||
if (packet.Length != PacketBytes)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
long window = WindowOf(nowMs);
|
||||
Span<byte> expected = stackalloc byte[TagBytes];
|
||||
bool ok = false;
|
||||
for (long w = window - 1; w <= window + 1; w++)
|
||||
{
|
||||
Tag(key, w, packet[..NonceBytes], expected);
|
||||
ok |= CryptographicOperations.FixedTimeEquals(expected, packet[NonceBytes..]);
|
||||
}
|
||||
|
||||
return ok;
|
||||
}
|
||||
|
||||
private static long WindowOf(long nowMs) => nowMs / 1000 / WindowSeconds;
|
||||
|
||||
private static void Tag(ReadOnlySpan<byte> key, long window, ReadOnlySpan<byte> nonce, Span<byte> destination)
|
||||
{
|
||||
Span<byte> data = stackalloc byte[8 + NonceBytes];
|
||||
BinaryPrimitives.WriteInt64BigEndian(data, window);
|
||||
nonce.CopyTo(data[8..]);
|
||||
Span<byte> full = stackalloc byte[32];
|
||||
HMACSHA256.HashData(key, data, full);
|
||||
full[..TagBytes].CopyTo(destination);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Net;
|
||||
using System.Security.Cryptography;
|
||||
|
||||
namespace Fleet.Identity;
|
||||
|
||||
/// <summary>
|
||||
/// 监听方的敲门关卡:新连接的前 32 字节必须是名册里某个成员的敲门包,否则不写任何字节、直接断开。
|
||||
/// 每个来源地址限速(陌生人用随机数据也要消耗 HMAC 计算),nonce 在时间窗内去重(防重放)。
|
||||
/// 名册由调用方提供(对端公钥列表);配对密钥按公钥缓存,名册变了自然只算新的。
|
||||
/// </summary>
|
||||
public sealed class KnockGate
|
||||
{
|
||||
private static readonly TimeSpan s_readTimeout = TimeSpan.FromSeconds(3);
|
||||
|
||||
private readonly NodeIdentity _own;
|
||||
private readonly Func<IReadOnlyList<byte[]>> _peers;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly int _attemptsPerPeriod;
|
||||
private readonly long _periodMs;
|
||||
private readonly ConcurrentDictionary<string, byte[]> _keys = new();
|
||||
private readonly object _gate = new();
|
||||
private readonly Dictionary<string, long> _seenNonces = [];
|
||||
private readonly Dictionary<string, (long WindowStartMs, int Count)> _sources = [];
|
||||
private long _lastSweepMs;
|
||||
|
||||
public KnockGate(NodeIdentity own, Func<IReadOnlyList<byte[]>> peers, TimeProvider time, int attemptsPerPeriod = 300, TimeSpan? period = null)
|
||||
{
|
||||
_own = own;
|
||||
_peers = peers;
|
||||
_time = time;
|
||||
_attemptsPerPeriod = attemptsPerPeriod;
|
||||
_periodMs = (long)(period ?? TimeSpan.FromSeconds(10)).TotalMilliseconds;
|
||||
}
|
||||
|
||||
/// <summary>读 32 字节并核对。通过返回 true,流从敲门包之后继续(TLS 或明文首帧);不通过返回 false,调用方直接关闭连接,不要写任何东西。</summary>
|
||||
public async ValueTask<bool> AdmitAsync(Stream stream, IPAddress? source, CancellationToken ct)
|
||||
{
|
||||
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||||
if (!TryCountSource(source, now))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
byte[] packet = new byte[Knock.PacketBytes];
|
||||
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
timeout.CancelAfter(s_readTimeout);
|
||||
try
|
||||
{
|
||||
await stream.ReadExactlyAsync(packet, timeout.Token);
|
||||
}
|
||||
catch (Exception ex) when (ex is EndOfStreamException or OperationCanceledException or IOException or ObjectDisposedException)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return Accept(packet, now);
|
||||
}
|
||||
|
||||
/// <summary>核对一个敲门包(不读流):名册里有成员的密钥对得上,并且 nonce 没用过。</summary>
|
||||
public bool Accept(ReadOnlySpan<byte> packet, long nowMs)
|
||||
{
|
||||
bool matched = false;
|
||||
foreach (byte[] peer in _peers())
|
||||
{
|
||||
byte[] key = _keys.GetOrAdd(Convert.ToBase64String(peer), _ => Knock.DeriveKey(_own, peer));
|
||||
matched |= Knock.Check(key, packet, nowMs);
|
||||
}
|
||||
|
||||
if (!matched)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
string nonce = Convert.ToHexString(packet[..Knock.NonceBytes]);
|
||||
lock (_gate)
|
||||
{
|
||||
Sweep(nowMs);
|
||||
return _seenNonces.TryAdd(nonce, nowMs + (Knock.WindowSeconds * 3 * 1000L));
|
||||
}
|
||||
}
|
||||
|
||||
private bool TryCountSource(IPAddress? source, long nowMs)
|
||||
{
|
||||
string key = source is null ? "" : source.IsIPv4MappedToIPv6 ? source.MapToIPv4().ToString() : source.ToString();
|
||||
lock (_gate)
|
||||
{
|
||||
Sweep(nowMs);
|
||||
(long start, int count) = _sources.TryGetValue(key, out var entry) && nowMs - entry.WindowStartMs < _periodMs ? entry : (nowMs, 0);
|
||||
_sources[key] = (start, count + 1);
|
||||
return count < _attemptsPerPeriod;
|
||||
}
|
||||
}
|
||||
|
||||
private void Sweep(long nowMs)
|
||||
{
|
||||
if (nowMs - _lastSweepMs < 5000 && _seenNonces.Count < 4096 && _sources.Count < 4096)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_lastSweepMs = nowMs;
|
||||
foreach (string nonce in _seenNonces.Where(p => p.Value <= nowMs).Select(p => p.Key).ToList())
|
||||
{
|
||||
_seenNonces.Remove(nonce);
|
||||
}
|
||||
|
||||
foreach (string source in _sources.Where(p => nowMs - p.Value.WindowStartMs >= _periodMs).Select(p => p.Key).ToList())
|
||||
{
|
||||
_sources.Remove(source);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -41,6 +41,16 @@ public sealed class NodeIdentity : IDisposable
|
||||
|
||||
public byte[] ExportPkcs8() => _key.ExportPkcs8PrivateKey();
|
||||
|
||||
/// <summary>与对端公钥(SPKI)做静态 ECDH,返回原始共享秘密。只用于派生对称密钥(敲门),必须再经 KDF,不能直接当密钥用。</summary>
|
||||
public byte[] AgreeWith(ReadOnlySpan<byte> peerSpki)
|
||||
{
|
||||
using ECDiffieHellman own = ECDiffieHellman.Create();
|
||||
own.ImportPkcs8PrivateKey(_key.ExportPkcs8PrivateKey(), out _);
|
||||
using ECDiffieHellman peer = ECDiffieHellman.Create();
|
||||
peer.ImportSubjectPublicKeyInfo(peerSpki, out _);
|
||||
return own.DeriveRawSecretAgreement(peer.PublicKey);
|
||||
}
|
||||
|
||||
public byte[] Sign(string purpose, params string[] fields) =>
|
||||
_key.SignData(Encoding.UTF8.GetBytes(SignatureString.Build(purpose, fields)), HashAlgorithmName.SHA256,
|
||||
DSASignatureFormat.IeeeP1363FixedFieldConcatenation);
|
||||
|
||||
@@ -22,6 +22,9 @@ public sealed class PeerListenerOptions
|
||||
/// <summary>按对端 ID 查找允许连接的公钥(SPKI)。不在列表里的对端一律拒绝(期望状态里的 peers)。</summary>
|
||||
public required Func<string, byte[]?> LookupPeerKey { get; init; }
|
||||
|
||||
/// <summary>预握手敲门:每个连接先过这一关(名册里的 Probe 才算得出敲门包),不通过就直接断开,什么都不写。</summary>
|
||||
public required KnockGate Knock { get; init; }
|
||||
|
||||
public int MaxFrameBytes { get; init; } = 8 << 20;
|
||||
|
||||
public int MaxSessions { get; init; } = 8;
|
||||
@@ -98,7 +101,13 @@ public sealed class PeerListener
|
||||
PeerSockets.EnableKeepAlive(client.Client);
|
||||
using var timeout = new CancellationTokenSource(_options.HandshakeTimeout, _time);
|
||||
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token);
|
||||
var tls = new SslStream(client.GetStream(), leaveInnerStreamOpen: false);
|
||||
Stream stream = client.GetStream();
|
||||
if (!await _options.Knock.AdmitAsync(stream, (client.Client.RemoteEndPoint as IPEndPoint)?.Address, linked.Token))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var tls = new SslStream(stream, leaveInnerStreamOpen: false);
|
||||
await tls.AuthenticateAsServerAsync(
|
||||
new SslServerAuthenticationOptions { ServerCertificate = _options.Certificate, ClientCertificateRequired = false, EnabledSslProtocols = System.Security.Authentication.SslProtocols.Tls12 | System.Security.Authentication.SslProtocols.Tls13 },
|
||||
linked.Token);
|
||||
@@ -222,8 +231,10 @@ internal static class PeerSockets
|
||||
public static class PeerConnector
|
||||
{
|
||||
public static async Task<(PeerSessionInfo Info, TlsFrameTransport Transport)> ConnectAsync(
|
||||
string host, int port, byte[] expectedSpkiSha256, NodeIdentity identity, string peerId, string? version, int maxFrameBytes, TimeProvider time, CancellationToken ct)
|
||||
string host, int port, byte[] agentPublicKey, NodeIdentity identity, string peerId, string? version, int maxFrameBytes, TimeProvider time, CancellationToken ct)
|
||||
{
|
||||
// 证书按公钥的指纹固定;敲门密钥由双方身份密钥派生
|
||||
byte[] expectedSpkiSha256 = NodeIdentity.Fingerprint(agentPublicKey);
|
||||
TcpClient client = IPAddress.TryParse(host, out IPAddress? address) ? new TcpClient(address.AddressFamily) : new TcpClient();
|
||||
client.NoDelay = true;
|
||||
PeerSockets.EnableKeepAlive(client.Client);
|
||||
@@ -232,8 +243,10 @@ public static class PeerConnector
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10), time);
|
||||
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token);
|
||||
await client.ConnectAsync(host, port, linked.Token);
|
||||
Stream stream = client.GetStream();
|
||||
await stream.WriteAsync(Knock.Build(Knock.DeriveKey(identity, agentPublicKey), time.GetUtcNow().ToUnixTimeMilliseconds()), linked.Token);
|
||||
var tls = new SslStream(
|
||||
client.GetStream(), leaveInnerStreamOpen: false,
|
||||
stream, leaveInnerStreamOpen: false,
|
||||
(_, certificate, _, _) => certificate is not null && SpkiMatches(certificate, expectedSpkiSha256));
|
||||
await tls.AuthenticateAsClientAsync(new SslClientAuthenticationOptions { TargetHost = host }, linked.Token);
|
||||
var transport = new TlsFrameTransport(tls, maxFrameBytes);
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
using System.Net;
|
||||
using Fleet.Identity;
|
||||
|
||||
namespace Fleet.Tests;
|
||||
|
||||
public sealed class KnockTests
|
||||
{
|
||||
private sealed class Clock : TimeProvider
|
||||
{
|
||||
public DateTimeOffset Now { get; set; } = new(2026, 10, 7, 12, 0, 0, TimeSpan.Zero);
|
||||
|
||||
public override DateTimeOffset GetUtcNow() => Now;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void PairKey_IsTheSameOnBothSides_AndDiffersPerPair()
|
||||
{
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
using NodeIdentity agent = NodeIdentity.Generate();
|
||||
using NodeIdentity other = NodeIdentity.Generate();
|
||||
byte[] fromProbe = Knock.DeriveKey(probe, agent.PublicKeySpki);
|
||||
byte[] fromAgent = Knock.DeriveKey(agent, probe.PublicKeySpki);
|
||||
Assert.Equal(fromProbe, fromAgent);
|
||||
Assert.Equal(32, fromProbe.Length);
|
||||
Assert.NotEqual(fromProbe, Knock.DeriveKey(other, agent.PublicKeySpki));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Packet_IsAcceptedWithinTheWindowAndOneWindowOfSkew_ButNotBeyond()
|
||||
{
|
||||
byte[] key = new byte[32];
|
||||
long now = new DateTimeOffset(2026, 10, 7, 12, 0, 15, TimeSpan.Zero).ToUnixTimeMilliseconds();
|
||||
byte[] packet = Knock.Build(key, now);
|
||||
Assert.Equal(Knock.PacketBytes, packet.Length);
|
||||
Assert.True(Knock.Check(key, packet, now));
|
||||
Assert.True(Knock.Check(key, packet, now + (Knock.WindowSeconds * 1000)));
|
||||
Assert.True(Knock.Check(key, packet, now - (Knock.WindowSeconds * 1000)));
|
||||
Assert.False(Knock.Check(key, packet, now + (3 * Knock.WindowSeconds * 1000)));
|
||||
Assert.False(Knock.Check(key, packet, now - (3 * Knock.WindowSeconds * 1000)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Packet_WithAWrongKeyATamperedByteOrTheWrongLength_IsRejected()
|
||||
{
|
||||
byte[] key = new byte[32];
|
||||
long now = 1_700_000_000_000;
|
||||
byte[] packet = Knock.Build(key, now);
|
||||
Assert.False(Knock.Check(new byte[32] { 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }, packet, now));
|
||||
foreach (int index in new[] { 0, 15, 16, 31 })
|
||||
{
|
||||
byte[] tampered = (byte[])packet.Clone();
|
||||
tampered[index] ^= 0x01;
|
||||
Assert.False(Knock.Check(key, tampered, now), $"byte {index}");
|
||||
}
|
||||
|
||||
Assert.False(Knock.Check(key, packet.AsSpan(0, 31), now));
|
||||
Assert.False(Knock.Check(key, [], now));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Gate_AdmitsRosterMembersOnce_AndRejectsReplaysStrangersAndExpiredPackets()
|
||||
{
|
||||
using NodeIdentity agent = NodeIdentity.Generate();
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
using NodeIdentity stranger = NodeIdentity.Generate();
|
||||
var roster = new List<byte[]> { probe.PublicKeySpki };
|
||||
var gate = new KnockGate(agent, () => roster, new Clock());
|
||||
long now = 1_700_000_000_000;
|
||||
|
||||
byte[] packet = Knock.Build(Knock.DeriveKey(probe, agent.PublicKeySpki), now);
|
||||
Assert.True(gate.Accept(packet, now));
|
||||
Assert.False(gate.Accept(packet, now), "the same nonce cannot be used twice");
|
||||
Assert.False(gate.Accept(Knock.Build(Knock.DeriveKey(stranger, agent.PublicKeySpki), now), now), "not on the roster");
|
||||
Assert.False(gate.Accept(Knock.Build(Knock.DeriveKey(probe, agent.PublicKeySpki), now - 10 * 60_000), now), "expired");
|
||||
|
||||
// 名册变了:新成员马上能进,被移除的马上进不来
|
||||
using NodeIdentity second = NodeIdentity.Generate();
|
||||
roster.Add(second.PublicKeySpki);
|
||||
Assert.True(gate.Accept(Knock.Build(Knock.DeriveKey(second, agent.PublicKeySpki), now), now));
|
||||
roster.RemoveAt(0);
|
||||
Assert.False(gate.Accept(Knock.Build(Knock.DeriveKey(probe, agent.PublicKeySpki), now), now));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Gate_RateLimitsEachSource_AndReadsAtMost32Bytes()
|
||||
{
|
||||
using NodeIdentity agent = NodeIdentity.Generate();
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
var clock = new Clock();
|
||||
var gate = new KnockGate(agent, () => [probe.PublicKeySpki], clock, attemptsPerPeriod: 3);
|
||||
IPAddress noisy = IPAddress.Parse("203.0.113.9");
|
||||
IPAddress quiet = IPAddress.Parse("203.0.113.10");
|
||||
byte[] key = Knock.DeriveKey(probe, agent.PublicKeySpki);
|
||||
|
||||
for (int i = 0; i < 3; i++)
|
||||
{
|
||||
Assert.False(await gate.AdmitAsync(new MemoryStream(new byte[Knock.PacketBytes]), noisy, default));
|
||||
}
|
||||
|
||||
// 超过限额:连有效的敲门包也不读、不算
|
||||
byte[] valid = Knock.Build(key, clock.Now.ToUnixTimeMilliseconds());
|
||||
Assert.False(await gate.AdmitAsync(new MemoryStream(valid), noisy, default));
|
||||
Assert.True(await gate.AdmitAsync(new MemoryStream(valid), quiet, default));
|
||||
|
||||
// 过了统计周期,恢复
|
||||
clock.Now += TimeSpan.FromSeconds(11);
|
||||
Assert.True(await gate.AdmitAsync(new MemoryStream(Knock.Build(key, clock.Now.ToUnixTimeMilliseconds())), noisy, default));
|
||||
|
||||
// 流在 32 字节之前就结束:拒绝,不抛异常
|
||||
Assert.False(await gate.AdmitAsync(new MemoryStream(new byte[10]), quiet, default));
|
||||
}
|
||||
}
|
||||
@@ -280,12 +280,20 @@ public sealed class LinkTests
|
||||
private sealed class PeerFixture : IAsyncDisposable
|
||||
{
|
||||
private readonly CancellationTokenSource _stop = new();
|
||||
private readonly NodeIdentity _identity;
|
||||
|
||||
public PeerFixture(Func<string, byte[]?> lookup, Func<PeerSessionInfo, IRpcTransport, CancellationToken, Task>? onSession = null)
|
||||
{
|
||||
Certificate = NodeCertificate.LoadPfx(NodeCertificate.CreateSelfSignedPfx("agent", TimeSpan.FromDays(30), TimeProvider.System));
|
||||
_identity = NodeIdentity.FromPkcs8(Certificate.GetECDsaPrivateKey()!.ExportPkcs8PrivateKey());
|
||||
Listener = new PeerListener(
|
||||
new PeerListenerOptions { Listen = new IPEndPoint(IPAddress.Loopback, 0), Certificate = Certificate, LookupPeerKey = lookup },
|
||||
new PeerListenerOptions
|
||||
{
|
||||
Listen = new IPEndPoint(IPAddress.Loopback, 0),
|
||||
Certificate = Certificate,
|
||||
LookupPeerKey = lookup,
|
||||
Knock = new KnockGate(_identity, () => KnockPeers, TimeProvider.System),
|
||||
},
|
||||
onSession ?? ((_, transport, ct) => KeepOpenAsync(transport, ct)),
|
||||
NullNodeLog.Instance, TimeProvider.System);
|
||||
Task = Listener.RunAsync(_stop.Token);
|
||||
@@ -299,10 +307,17 @@ public sealed class LinkTests
|
||||
|
||||
public byte[] Spki => NodeCertificate.SpkiFingerprint(Certificate);
|
||||
|
||||
/// <summary>Agent 的 TLS 证书公钥:Probe 用它固定证书、派生敲门密钥。</summary>
|
||||
public byte[] PublicKey => Certificate.PublicKey.ExportSubjectPublicKeyInfo();
|
||||
|
||||
/// <summary>名册:能敲开门的 Probe 公钥。</summary>
|
||||
public List<byte[]> KnockPeers { get; } = [];
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _stop.CancelAsync();
|
||||
await Task;
|
||||
_identity.Dispose();
|
||||
Certificate.Dispose();
|
||||
_stop.Dispose();
|
||||
}
|
||||
@@ -325,9 +340,10 @@ public sealed class LinkTests
|
||||
seen.TrySetResult(info);
|
||||
return Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, CancellationToken.None);
|
||||
});
|
||||
agent.KnockPeers.Add(probe.PublicKeySpki);
|
||||
await agent.Listener.Started.Task;
|
||||
|
||||
(PeerSessionInfo client, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
(PeerSessionInfo client, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
await using (transport)
|
||||
{
|
||||
PeerSessionInfo server = await seen.Task.WaitAsync(TimeSpan.FromSeconds(5));
|
||||
@@ -343,17 +359,18 @@ public sealed class LinkTests
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
using NodeIdentity impostor = NodeIdentity.Generate();
|
||||
await using var agent = new PeerFixture(id => id == "prb_1" ? probe.PublicKeySpki : null);
|
||||
agent.KnockPeers.AddRange([probe.PublicKeySpki, impostor.PublicKeySpki]); // 敲门通过之后,名册外的 ID 和伪造的签名仍然被拒绝
|
||||
await agent.Listener.Started.Task;
|
||||
|
||||
InvalidOperationException unknown = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
|
||||
await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, probe, "prb_unknown", null, 1 << 20, TimeProvider.System, default));
|
||||
await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_unknown", null, 1 << 20, TimeProvider.System, default));
|
||||
InvalidOperationException forged = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
|
||||
await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, impostor, "prb_1", null, 1 << 20, TimeProvider.System, default));
|
||||
await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, impostor, "prb_1", null, 1 << 20, TimeProvider.System, default));
|
||||
Assert.Equal(unknown.Message, forged.Message);
|
||||
Assert.Contains("fleet_unauthorized", forged.Message);
|
||||
|
||||
// 正主仍然可以连接
|
||||
(_, TlsFrameTransport ok) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
(_, TlsFrameTransport ok) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
await ok.DisposeAsync();
|
||||
}
|
||||
|
||||
@@ -362,11 +379,13 @@ public sealed class LinkTests
|
||||
{
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
await using var agent = new PeerFixture(_ => probe.PublicKeySpki);
|
||||
agent.KnockPeers.Add(probe.PublicKeySpki);
|
||||
await agent.Listener.Started.Task;
|
||||
|
||||
byte[] wrong = SHA256.HashData("another certificate"u8);
|
||||
await Assert.ThrowsAnyAsync<AuthenticationException>(async () =>
|
||||
await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, wrong, probe, "prb_1", null, 1 << 20, TimeProvider.System, default));
|
||||
// 公钥不对:敲门密钥对不上,Agent 直接断开(拿到的不是证书,连不是它的证书都看不到)
|
||||
using NodeIdentity another = NodeIdentity.Generate();
|
||||
await Assert.ThrowsAnyAsync<Exception>(async () =>
|
||||
await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, another.PublicKeySpki, probe, "prb_1", null, 1 << 20, TimeProvider.System, default));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -382,9 +401,10 @@ public sealed class LinkTests
|
||||
await using var session = new RpcSession<string>(transport, router, info.PeerId, TimeProvider.System);
|
||||
await session.RunAsync(ct);
|
||||
});
|
||||
agent.KnockPeers.Add(probe.PublicKeySpki);
|
||||
await agent.Listener.Started.Task;
|
||||
|
||||
(_, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
(_, TlsFrameTransport transport) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
await using var client = new RpcSession<string>(transport, new RpcRouter<string>(RpcSide.Dialer, table), "probe", TimeProvider.System);
|
||||
Task run = client.RunAsync(default);
|
||||
Result<Number> result = await client.RequestAsync("agent.echo", new Number { Value = 41 }, LinkJson.Default.Number, LinkJson.Default.Number, default);
|
||||
@@ -398,6 +418,7 @@ public sealed class LinkTests
|
||||
{
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
await using var agent = new PeerFixture(_ => probe.PublicKeySpki);
|
||||
agent.KnockPeers.Add(probe.PublicKeySpki);
|
||||
await agent.Listener.Started.Task;
|
||||
|
||||
// 不说 TLS 的连接:握手失败,监听继续
|
||||
@@ -407,7 +428,45 @@ public sealed class LinkTests
|
||||
await garbage.GetStream().WriteAsync("GET / HTTP/1.1\r\n\r\n"u8.ToArray());
|
||||
}
|
||||
|
||||
(_, TlsFrameTransport ok) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.Spki, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
(_, TlsFrameTransport ok) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
await ok.DisposeAsync();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PeerListener_WritesNothingToStrangers_NotEvenATlsServerHello()
|
||||
{
|
||||
using NodeIdentity probe = NodeIdentity.Generate();
|
||||
await using var agent = new PeerFixture(_ => probe.PublicKeySpki);
|
||||
agent.KnockPeers.Add(probe.PublicKeySpki);
|
||||
await agent.Listener.Started.Task;
|
||||
|
||||
// 一个像样的 TLS ClientHello 开头、随机数据、什么都不发:都拿不到任何字节(连接被直接关闭)
|
||||
foreach (byte[] payload in new[] { new byte[] { 0x16, 0x03, 0x01, 0x00, 0x05, 0x01, 0x00, 0x00, 0x01, 0x00 }.Concat(RandomNumberGenerator.GetBytes(64)).ToArray(), RandomNumberGenerator.GetBytes(32), [] })
|
||||
{
|
||||
using var scanner = new TcpClient();
|
||||
await scanner.ConnectAsync(IPAddress.Loopback, agent.Listener.Port);
|
||||
NetworkStream stream = scanner.GetStream();
|
||||
if (payload.Length > 0)
|
||||
{
|
||||
await stream.WriteAsync(payload);
|
||||
}
|
||||
|
||||
byte[] buffer = new byte[64];
|
||||
int read;
|
||||
try
|
||||
{
|
||||
read = await stream.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10));
|
||||
}
|
||||
catch (IOException)
|
||||
{
|
||||
read = 0; // 连接被重置也是「什么都没收到」
|
||||
}
|
||||
|
||||
Assert.Equal(0, read);
|
||||
}
|
||||
|
||||
// 名册里的 Probe 仍然进得来
|
||||
(_, TlsFrameTransport ok) = await PeerConnector.ConnectAsync("127.0.0.1", agent.Listener.Port, agent.PublicKey, probe, "prb_1", null, 1 << 20, TimeProvider.System, default);
|
||||
await ok.DisposeAsync();
|
||||
}
|
||||
|
||||
|
||||
@@ -89,8 +89,9 @@ public sealed class AgentConnection
|
||||
throw new SocketException((int)SocketError.HostNotFound);
|
||||
}
|
||||
|
||||
byte[] agentKey = Convert.FromBase64String(Endpoint.PublicKey);
|
||||
(PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(
|
||||
address.ToString(), Endpoint.TcpPort, Convert.FromBase64String(Endpoint.Spki), _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, ct);
|
||||
address.ToString(), Endpoint.TcpPort, agentKey, _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, ct);
|
||||
var remote = (IPEndPoint)info.Remote!;
|
||||
var data = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key);
|
||||
using var udp = new UdpMux(new SingleSession(data), null, _env.Clock, new DataPlaneCounters(), _env.Log, remote.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, []);
|
||||
@@ -105,7 +106,8 @@ public sealed class AgentConnection
|
||||
Session = () => data,
|
||||
Udp = udp,
|
||||
PeerAddress = remote.Address,
|
||||
PeerSpkiSha256 = Convert.FromBase64String(Endpoint.Spki),
|
||||
PeerSpkiSha256 = NodeIdentity.Fingerprint(agentKey),
|
||||
KnockKey = Knock.DeriveKey(_env.Identity, agentKey),
|
||||
Fragments = shipper,
|
||||
MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
|
||||
MaxBulkMbps = _env.Limits.MaxTestMbps,
|
||||
|
||||
@@ -72,12 +72,12 @@ public sealed class AgentDirectory
|
||||
{
|
||||
if (!string.IsNullOrWhiteSpace(agent.Host))
|
||||
{
|
||||
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Host, TcpPort = agent.TcpPort, Spki = agent.Spki });
|
||||
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Host, TcpPort = agent.TcpPort, PublicKey = agent.PublicKey });
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(agent.Ipv6Host))
|
||||
{
|
||||
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Ipv6Host, TcpPort = agent.TcpPort, Spki = agent.Spki, IpVersion = IpVersions.V6 });
|
||||
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Ipv6Host, TcpPort = agent.TcpPort, PublicKey = agent.PublicKey, IpVersion = IpVersions.V6 });
|
||||
}
|
||||
}
|
||||
lock (_gate)
|
||||
@@ -108,5 +108,5 @@ public sealed class AgentDirectory
|
||||
a.Count == b.Count && a.Zip(b).All(p => Equal(p.First, p.Second));
|
||||
|
||||
public static bool Equal(KnownAgent a, KnownAgent b) =>
|
||||
a.AgentId == b.AgentId && a.IpVersion == b.IpVersion && a.Host == b.Host && a.TcpPort == b.TcpPort && a.Spki == b.Spki;
|
||||
a.AgentId == b.AgentId && a.IpVersion == b.IpVersion && a.Host == b.Host && a.TcpPort == b.TcpPort && a.PublicKey == b.PublicKey;
|
||||
}
|
||||
@@ -48,9 +48,9 @@ public sealed class KnownAgent
|
||||
[JsonPropertyName("tcp_port")]
|
||||
public int TcpPort { get; set; }
|
||||
|
||||
/// <summary>Agent 的 TLS 证书指纹(SHA-256,base64)。</summary>
|
||||
[JsonPropertyName("spki")]
|
||||
public string Spki { get; set; } = "";
|
||||
/// <summary>Agent 的 TLS 证书公钥(SPKI,base64):证书按它的 SHA-256 指纹固定,敲门密钥由它派生。</summary>
|
||||
[JsonPropertyName("public_key")]
|
||||
public string PublicKey { get; set; } = "";
|
||||
}
|
||||
|
||||
public sealed class ProbeLimits
|
||||
|
||||
@@ -58,7 +58,7 @@ public sealed class ProbeTests
|
||||
private static ProbeAgentList List(long version, params string[] ids) => new()
|
||||
{
|
||||
Version = version,
|
||||
Agents = [.. ids.Select(id => new AgentEndpoint { AgentId = new NodeId(id), Host = id + ".example", TcpPort = 7443, Spki = "c3BraQ==" })],
|
||||
Agents = [.. ids.Select(id => new AgentEndpoint { AgentId = new NodeId(id), Host = id + ".example", TcpPort = 7443, PublicKey = "c3BraQ==" })],
|
||||
};
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -217,6 +217,9 @@ public sealed class BlockEnv
|
||||
/// <summary>拨号方:对端 TLS 证书的 SPKI 指纹;null 表示不校验证书(目标任务,目标是任意地址)。</summary>
|
||||
public byte[]? PeerSpkiSha256 { get; init; }
|
||||
|
||||
/// <summary>拨号方:和对端(Agent)的配对敲门密钥;每条 TCP 连接建立后先发敲门包。null 表示对端不要求敲门(目标任务的任意地址)。</summary>
|
||||
public byte[]? KnockKey { get; init; }
|
||||
|
||||
/// <summary>拨号方:列块出口。</summary>
|
||||
public IFragmentSink? Fragments { get; init; }
|
||||
|
||||
@@ -241,6 +244,7 @@ public sealed class BlockEnv
|
||||
TcpListen = TcpListen,
|
||||
PeerAddress = PeerAddress,
|
||||
PeerSpkiSha256 = PeerSpkiSha256,
|
||||
KnockKey = KnockKey,
|
||||
Fragments = Fragments,
|
||||
MaxBulkMbps = MaxBulkMbps,
|
||||
MaxPacketsPerSecond = MaxPacketsPerSecond,
|
||||
|
||||
@@ -6,5 +6,6 @@
|
||||
<InternalsVisibleTo Include="MoShiro.Measurement.Tests" />
|
||||
<AdditionalFiles Include="$(MSBuildThisFileDirectory)../../BannedSymbols.Node.txt" />
|
||||
<ProjectReference Include="..\MoShiro.Protocol\MoShiro.Protocol.csproj" />
|
||||
<ProjectReference Include="..\..\fleet\src\Fleet.Identity\Fleet.Identity.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
@@ -5,6 +5,7 @@ using System.Net.Sockets;
|
||||
using System.Security.Authentication;
|
||||
using System.Security.Cryptography;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using Fleet.Identity;
|
||||
using MoShiro.Protocol;
|
||||
|
||||
namespace MoShiro.Measurement;
|
||||
@@ -46,6 +47,12 @@ public sealed class TcpLink : IDisposable
|
||||
await client.ConnectAsync(env.PeerAddress!, port, ct);
|
||||
long tcpUs = env.Clock.NowUs() - start;
|
||||
Stream stream = client.GetStream();
|
||||
if (env.KnockKey is { } knockKey)
|
||||
{
|
||||
// 预握手敲门:监听方不认就不会回任何字节
|
||||
await stream.WriteAsync(Knock.Build(knockKey, env.Time.GetUtcNow().ToUnixTimeMilliseconds()), ct);
|
||||
}
|
||||
|
||||
long? tlsUs = null;
|
||||
if (tls)
|
||||
{
|
||||
@@ -115,7 +122,7 @@ public sealed class TcpConnection
|
||||
public delegate Task TcpResponder(TcpConnection connection, CancellationToken ct);
|
||||
|
||||
/// <summary>
|
||||
/// 监听方的 TCP 数据面:每个端口一个监听器。连接的第一个字节是 0x16 就先做 TLS(节点自己的证书),否则直接是明文。
|
||||
/// 监听方的 TCP 数据面:每个端口一个监听器。连接先过预握手敲门,然后第一个字节是 0x16 就先做 TLS(节点自己的证书),否则直接是明文。
|
||||
/// 随后是认证首帧(<c>"MN" | ver | session_id | stream_id | nonce | mac</c>,随机数只能用一次),通过之后交给按 <c>stream_id</c> 登记的处理函数。
|
||||
/// 首帧在限定时间内没有到达或者校验失败,直接关闭连接并计数。
|
||||
/// </summary>
|
||||
@@ -128,11 +135,13 @@ public sealed class TcpAcceptor
|
||||
private readonly X509Certificate2 _certificate;
|
||||
private readonly IDataSessions _sessions;
|
||||
private readonly DataPlaneCounters _counters;
|
||||
private readonly KnockGate _knock;
|
||||
private readonly ConcurrentDictionary<uint, TcpResponder> _handlers = new();
|
||||
private readonly List<TcpListener> _listeners = [];
|
||||
|
||||
public TcpAcceptor(IPAddress bind, IReadOnlyList<int> ports, X509Certificate2 certificate, IDataSessions sessions, DataPlaneCounters counters)
|
||||
public TcpAcceptor(IPAddress bind, IReadOnlyList<int> ports, X509Certificate2 certificate, IDataSessions sessions, DataPlaneCounters counters, KnockGate knock)
|
||||
{
|
||||
_knock = knock;
|
||||
_bind = bind;
|
||||
_ports = ports;
|
||||
_certificate = certificate;
|
||||
@@ -214,6 +223,13 @@ public sealed class TcpAcceptor
|
||||
using var authCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
authCts.CancelAfter(s_authTimeout);
|
||||
|
||||
// 预握手敲门:名册里的 Probe 才算得出敲门包;不通过就不写任何字节、直接断开
|
||||
if (!await _knock.AdmitAsync(stream, (client.Client.RemoteEndPoint as IPEndPoint)?.Address, authCts.Token))
|
||||
{
|
||||
_counters.DropInvalid();
|
||||
return;
|
||||
}
|
||||
|
||||
byte[] first = new byte[1];
|
||||
if (await stream.ReadAsync(first, authCts.Token) != 1)
|
||||
{
|
||||
|
||||
@@ -149,9 +149,9 @@ public sealed class AgentEndpoint
|
||||
[JsonPropertyName("udp_ports")]
|
||||
public List<int> UdpPorts { get; set; } = [];
|
||||
|
||||
/// <summary>Agent 面向 Probe 的 TLS 证书指纹(SHA-256,base64)。</summary>
|
||||
[JsonPropertyName("spki")]
|
||||
public string Spki { get; set; } = "";
|
||||
/// <summary>Agent 面向 Probe 的 TLS 证书公钥(SPKI,base64):证书按它的 SHA-256 指纹固定,敲门密钥由它和 Probe 的身份密钥派生。</summary>
|
||||
[JsonPropertyName("public_key")]
|
||||
public string PublicKey { get; set; } = "";
|
||||
}
|
||||
|
||||
/// <summary>期望状态文档(Fleet state type = <c>probe_agents</c>):Backend 直接下发给每个 Probe 的 Agent 列表。</summary>
|
||||
@@ -323,8 +323,9 @@ public sealed class AgentHelloMeta
|
||||
[JsonPropertyName("platform")]
|
||||
public PlatformInfo? Platform { get; set; }
|
||||
|
||||
[JsonPropertyName("tls_spki")]
|
||||
public string? TlsSpki { get; set; }
|
||||
/// <summary>Agent 面向 Probe 的 TLS 证书公钥(SPKI,base64);面板据此得到证书指纹,并下发给 Probe 用来固定证书和派生敲门密钥。</summary>
|
||||
[JsonPropertyName("tls_pubkey")]
|
||||
public string? TlsPublicKey { get; set; }
|
||||
|
||||
/// <summary>本机的公网地址(IPv4 公网地址、IPv6 全球单播地址):面板据此自动填 <c>public_host</c> 和 <c>ipv6_host</c>,管理员手填的优先。</summary>
|
||||
[JsonPropertyName("addresses")]
|
||||
|
||||
@@ -157,7 +157,8 @@ public sealed partial class AgentService
|
||||
IReadOnlyList<AgentRecord> agents = await _agents.ListAsync(ct);
|
||||
long version = long.TryParse(await _settings.GetAsync(TaskPlanner.ProbeAgentsVersionKey, ct), out long v) ? v : 0;
|
||||
var result = new ProbeAgentList { Version = version };
|
||||
foreach (AgentRecord agent in agents.Where(a => a.Status == NodeStatus.Active && a.PublicHost.Length > 0 && links.IsLinked(probeId, a.Id.Value)))
|
||||
// 没有上报证书公钥的 Agent(旧版本)还不能被 Probe 连接:Probe 需要它来固定证书和派生敲门密钥
|
||||
foreach (AgentRecord agent in agents.Where(a => a.Status == NodeStatus.Active && a.PublicHost.Length > 0 && a.TlsPublicKey is { Length: > 0 } && links.IsLinked(probeId, a.Id.Value)))
|
||||
{
|
||||
result.Agents.Add(new AgentEndpoint
|
||||
{
|
||||
@@ -167,7 +168,7 @@ public sealed partial class AgentService
|
||||
Ipv6Host = agent.Ipv6Supported && agent.Ipv6Host.Length > 0 ? agent.Ipv6Host : null,
|
||||
TcpPort = agent.ProbeTcpPort,
|
||||
UdpPorts = [.. agent.ProbeUdpPorts],
|
||||
Spki = Convert.ToBase64String(agent.TlsSpki),
|
||||
PublicKey = Convert.ToBase64String(agent.TlsPublicKey!),
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -209,12 +209,19 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
|
||||
}
|
||||
}
|
||||
|
||||
// Agent 上报 TLS 证书的公钥:指纹由面板自己算,不信节点报的指纹
|
||||
byte[]? tlsKey = null;
|
||||
byte[]? spki = null;
|
||||
if (meta.TlsSpki is not null)
|
||||
if (meta.TlsPublicKey is not null)
|
||||
{
|
||||
try
|
||||
{
|
||||
spki = Convert.FromBase64String(meta.TlsSpki);
|
||||
byte[] candidate = Convert.FromBase64String(meta.TlsPublicKey);
|
||||
if (Fleet.Identity.NodeIdentity.IsValidPublicKey(candidate))
|
||||
{
|
||||
tlsKey = candidate;
|
||||
spki = Fleet.Identity.NodeIdentity.Fingerprint(candidate);
|
||||
}
|
||||
}
|
||||
catch (FormatException)
|
||||
{
|
||||
@@ -225,7 +232,7 @@ 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);
|
||||
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, meta.Addresses), ct);
|
||||
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts, meta.Ipv6Supported, ipv6Json, meta.Addresses, tlsKey), ct);
|
||||
|
||||
// 自动识别对外地址:Agent 上报的公网地址;手填的不动
|
||||
if (meta.Addresses is not null && await _agents.GetAsync(id, ct) is { } reported)
|
||||
@@ -240,7 +247,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
|
||||
await _releases.OfferToNodeAsync(id, ct);
|
||||
await _schedule.SendToNodeAsync(id, ct);
|
||||
if (before is not null && after is not null
|
||||
&& (!before.TlsSpki.AsSpan().SequenceEqual(after.TlsSpki) || before.ProbeTcpPort != after.ProbeTcpPort || !before.ProbeUdpPorts.SequenceEqual(after.ProbeUdpPorts) || before.Ipv6Supported != after.Ipv6Supported || before.PublicHost != after.PublicHost || before.Ipv6Host != after.Ipv6Host)
|
||||
&& (!before.TlsSpki.AsSpan().SequenceEqual(after.TlsSpki) || !(before.TlsPublicKey ?? []).AsSpan().SequenceEqual(after.TlsPublicKey ?? []) || before.ProbeTcpPort != after.ProbeTcpPort || !before.ProbeUdpPorts.SequenceEqual(after.ProbeUdpPorts) || before.Ipv6Supported != after.Ipv6Supported || before.PublicHost != after.PublicHost || before.Ipv6Host != after.Ipv6Host)
|
||||
&& after.Status == NodeStatus.Active)
|
||||
{
|
||||
await _service.AfterAgentSetChangedAsync(ct);
|
||||
|
||||
@@ -28,7 +28,8 @@ public sealed record AgentRecord(
|
||||
string? Ipv6DiagnosticsJson = null,
|
||||
bool PublicHostManual = false,
|
||||
bool Ipv6HostManual = false,
|
||||
IReadOnlyList<string>? ReportedAddresses = null);
|
||||
IReadOnlyList<string>? ReportedAddresses = null,
|
||||
byte[]? TlsPublicKey = null);
|
||||
|
||||
/// <summary>管理员可以修改的字段(merge-patch 语义:为 null 的字段不改)。</summary>
|
||||
public sealed class AgentPatch
|
||||
@@ -59,7 +60,8 @@ public sealed record AgentHello(
|
||||
IReadOnlyList<int>? ProbeUdpPorts,
|
||||
bool Ipv6Supported = false,
|
||||
string? Ipv6DiagnosticsJson = null,
|
||||
IReadOnlyList<string>? Addresses = null);
|
||||
IReadOnlyList<string>? Addresses = null,
|
||||
byte[]? TlsPublicKey = null);
|
||||
|
||||
/// <summary>注册之后 MoShiro 的补充:公开地址和端口。</summary>
|
||||
public sealed record AgentEnrollment(string PublicHost, int ProbeTcpPort, IReadOnlyList<int> ProbeUdpPorts, IReadOnlyList<string> Kinds);
|
||||
|
||||
@@ -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 = 9;
|
||||
public const int CurrentVersion = 10;
|
||||
|
||||
public static void Migrate(SqliteConnection connection)
|
||||
{
|
||||
@@ -131,6 +131,15 @@ internal static class Schema
|
||||
connection.Exec("PRAGMA user_version = 9");
|
||||
transaction.Commit();
|
||||
}
|
||||
|
||||
if (version < 10)
|
||||
{
|
||||
// 预握手敲门:Probe 需要 Agent 的证书公钥(不只是指纹)来派生配对密钥,由 Agent 在 hello 里上报
|
||||
using SqliteTransaction transaction = connection.BeginTransaction();
|
||||
connection.Exec("ALTER TABLE agents ADD COLUMN tls_pubkey BLOB");
|
||||
connection.Exec("PRAGMA user_version = 10");
|
||||
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, ipv6_diagnostics, public_host_manual, ipv6_host_manual, reported_addresses";
|
||||
"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";
|
||||
|
||||
private readonly SqliteDatabase _db;
|
||||
|
||||
@@ -123,11 +123,11 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
|
||||
_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), " +
|
||||
"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), ("@tcp", hello.ProbeTcpPort), ("@udp", hello.ProbeUdpPorts is null ? null : StorageJson.ToJson(hello.ProbeUdpPorts)), ("@v6", hello.Ipv6Supported ? 1 : 0), ("@diag", hello.Ipv6DiagnosticsJson),
|
||||
("@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);
|
||||
|
||||
@@ -266,7 +266,8 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
|
||||
r.NullableString(21),
|
||||
r.GetInt64(22) != 0,
|
||||
r.GetInt64(23) != 0,
|
||||
StorageJson.Strings(r.GetString(24)));
|
||||
StorageJson.Strings(r.GetString(24)),
|
||||
r.NullableBytes(25));
|
||||
|
||||
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);
|
||||
|
||||
@@ -1,4 +1,7 @@
|
||||
using System.Net;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using MoShiro.Backend.Application;
|
||||
using MoShiro.Protocol;
|
||||
using System.Text.Json.Nodes;
|
||||
using MoShiro.TestSupport;
|
||||
|
||||
@@ -87,4 +90,14 @@ public sealed class AddressTests
|
||||
await agent.HelloAsync();
|
||||
await BackendFixture.EventuallyAsync(async () => (await AgentAsync(id))["public_host"]!.GetValue<string>() == "1.0.0.1");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ProbeAgentList_CarriesTheCertificatePublicKeyTheAgentReported_ForPinningAndKnocking()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
AgentService service = _backend.Services.GetRequiredService<AgentService>();
|
||||
ProbeAgentList list = await service.ProbeAgentsAsync(_backend.DefaultProbeId, default);
|
||||
AgentEndpoint endpoint = Assert.Single(list.Agents, a => a.AgentId == agent.NodeId);
|
||||
Assert.Equal(Convert.ToBase64String(agent.TlsPublicKey), endpoint.PublicKey);
|
||||
}
|
||||
}
|
||||
@@ -95,6 +95,19 @@ public sealed class AgentEndToEndTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>读到 0 字节(对端关闭)或者连接被重置,都算「没收到任何东西」;收到数据返回字节数。</summary>
|
||||
internal static async Task<int> ReadOrResetAsync(Stream stream, byte[] buffer)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await stream.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10));
|
||||
}
|
||||
catch (IOException)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TcpDataPlane_ClosesConnectionsWithoutAValidFirstFrame()
|
||||
{
|
||||
@@ -106,7 +119,7 @@ public sealed class AgentEndToEndTests
|
||||
await client.ConnectAsync(System.Net.IPAddress.Loopback, agent.App.TcpPorts[0]);
|
||||
await client.GetStream().WriteAsync(new byte[TcpWire.AuthBytes]); // 全零:magic 不对
|
||||
byte[] buffer = new byte[16];
|
||||
Assert.Equal(0, await client.GetStream().ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10)));
|
||||
Assert.Equal(0, await ReadOrResetAsync(client.GetStream(), buffer)); // 连接被关闭:没有任何字节(Windows 上可能是 RST)
|
||||
await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid >= 1));
|
||||
}
|
||||
}
|
||||
@@ -146,7 +159,7 @@ public sealed class AgentEndToEndTests
|
||||
|
||||
// The probe process restarts: a new session with its own streams; sequence numbers start from 0 again
|
||||
await using ProbeSimulator second = first.Restarted();
|
||||
await second.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsSpki, agent.App.UdpPorts[0], agent.NodeId);
|
||||
await second.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsPublicKey, agent.App.UdpPorts[0], agent.NodeId);
|
||||
LineTaskSet set = await second.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.UdpPorts[0]));
|
||||
LineTask udp2 = set.Tasks.Single(t => t.Kind == "udp_echo");
|
||||
Assert.NotEqual(udp.StreamId, udp2.StreamId);
|
||||
@@ -208,11 +221,11 @@ public sealed class AgentEndToEndTests
|
||||
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
||||
await agent.CreateProbeAsync();
|
||||
await using var stranger = new ProbeSimulator("prb_stranger", Fleet.Identity.NodeIdentity.Generate());
|
||||
await Assert.ThrowsAnyAsync<Exception>(() => stranger.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsSpki, agent.App.UdpPorts[0], agent.NodeId));
|
||||
await Assert.ThrowsAnyAsync<Exception>(() => stranger.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsPublicKey, agent.App.UdpPorts[0], agent.NodeId));
|
||||
|
||||
// 证书指纹不对(中间人)也不会连上
|
||||
// 证书公钥不对(中间人)也不会连上
|
||||
await using ProbeSimulator probe = await agent.CreateProbeAsync();
|
||||
await Assert.ThrowsAnyAsync<Exception>(() => probe.ConnectAsync("127.0.0.1", agent.App.ControlPort, new byte[32], agent.App.UdpPorts[0], agent.NodeId));
|
||||
await Assert.ThrowsAnyAsync<Exception>(() => probe.ConnectAsync("127.0.0.1", agent.App.ControlPort, Fleet.Identity.NodeIdentity.Generate().PublicKeySpki, agent.App.UdpPorts[0], agent.NodeId));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
||||
@@ -39,7 +39,7 @@ public sealed class AgentHarness : IAsyncDisposable
|
||||
|
||||
public string DataDirectory => _directory;
|
||||
|
||||
public byte[] TlsSpki => Convert.FromBase64String(_identity.File.TlsSpki);
|
||||
public byte[] TlsPublicKey => _identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo();
|
||||
|
||||
public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? configure = null, string? name = null)
|
||||
{
|
||||
@@ -85,7 +85,7 @@ public sealed class AgentHarness : IAsyncDisposable
|
||||
public async Task<ProbeSimulator> ConnectProbeAsync()
|
||||
{
|
||||
ProbeSimulator probe = await CreateProbeAsync();
|
||||
await probe.ConnectAsync("127.0.0.1", App.ControlPort, TlsSpki, App.UdpPorts[0], NodeId);
|
||||
await probe.ConnectAsync("127.0.0.1", App.ControlPort, TlsPublicKey, App.UdpPorts[0], NodeId);
|
||||
return probe;
|
||||
}
|
||||
|
||||
|
||||
@@ -104,7 +104,8 @@ public sealed class ArchitectureTests
|
||||
Assert.Equal(["Fleet.Hosting", "Fleet.Link", "MoShiro.Measurement"], probe.ProjectReferences.Order().ToArray());
|
||||
|
||||
// 两个节点的共享库只依赖协议
|
||||
Assert.Equal(["MoShiro.Protocol"], Get("MoShiro.Measurement").ProjectReferences.ToArray());
|
||||
// 测量库在 TCP 数据面上做预握手敲门,所以直接引用 Fleet.Identity(节点侧库)
|
||||
Assert.Equal(["Fleet.Identity", "MoShiro.Protocol"], Get("MoShiro.Measurement").ProjectReferences.Order().ToArray());
|
||||
|
||||
// 节点一侧的库没有运行时依赖包,并使用节点禁用列表(不能用 HttpClient、Process、写文件;唯一的写文件豁免在 ConfigStore)
|
||||
foreach (string name in new[] { "Fleet.Link", "Fleet.Hosting", "MoShiro.Measurement", "MoShiro.Agent", "MoShiro.Probe" })
|
||||
|
||||
@@ -18,6 +18,7 @@ namespace MoShiro.TestSupport;
|
||||
public sealed class AgentSimulator : IAsyncDisposable
|
||||
{
|
||||
private readonly NodeIdentity _identity;
|
||||
private readonly NodeIdentity _tlsKey = NodeIdentity.Generate();
|
||||
private readonly ClientWebSocket _socket = new();
|
||||
private readonly ConcurrentDictionary<Guid, TaskCompletionSource<bool>> _acks = new();
|
||||
private readonly ConcurrentDictionary<long, TaskCompletionSource<RpcMessage>> _responses = new();
|
||||
@@ -71,6 +72,9 @@ public sealed class AgentSimulator : IAsyncDisposable
|
||||
|
||||
public byte[] PublicKey => _identity.PublicKeySpki;
|
||||
|
||||
/// <summary>hello 里上报的 TLS 证书公钥(Probe 用它固定证书、派生敲门密钥)。</summary>
|
||||
public byte[] TlsPublicKey => _tlsKey.PublicKeySpki;
|
||||
|
||||
/// <summary>用这个节点的私钥为认证随机数签名(可以冒充别的节点 ID:用来测试服务端的校验)。</summary>
|
||||
public byte[] SignAuth(NodeId claimedNode, string nonce) => _identity.Sign(SignaturePurpose.Auth, claimedNode.Value, nonce);
|
||||
|
||||
@@ -147,7 +151,7 @@ public sealed class AgentSimulator : IAsyncDisposable
|
||||
var meta = JsonSerializer.SerializeToElement(
|
||||
new AgentHelloMeta
|
||||
{
|
||||
TlsSpki = Convert.ToBase64String(RandomNumberGenerator.GetBytes(32)),
|
||||
TlsPublicKey = Convert.ToBase64String(_tlsKey.PublicKeySpki),
|
||||
Addresses = Addresses is null ? null : [.. Addresses],
|
||||
ProbeTcpPort = 7443,
|
||||
ProbeUdpPorts = [7443],
|
||||
|
||||
@@ -20,6 +20,7 @@ namespace MoShiro.TestSupport;
|
||||
public sealed class ProbeSimulator : IAsyncDisposable
|
||||
{
|
||||
private readonly NodeIdentity _identity;
|
||||
private byte[]? _agentKey;
|
||||
private readonly RpcRouter<ProbeSimulator> _router;
|
||||
private readonly ConcurrentDictionary<Guid, TaskCompletionSource> _acks = new();
|
||||
private readonly ConcurrentDictionary<uint, EchoCollector> _echoes = new();
|
||||
@@ -65,11 +66,12 @@ public sealed class ProbeSimulator : IAsyncDisposable
|
||||
|
||||
public long NowUs() => (TimeProvider.System.GetTimestamp() - _origin) * 1_000_000 / TimeProvider.System.TimestampFrequency;
|
||||
|
||||
public async Task ConnectAsync(string host, int controlPort, byte[] agentSpkiSha256, int udpPort, NodeId agentId, TimeProvider? time = null, CancellationToken ct = default)
|
||||
public async Task ConnectAsync(string host, int controlPort, byte[] agentPublicKey, int udpPort, NodeId agentId, TimeProvider? time = null, CancellationToken ct = default)
|
||||
{
|
||||
_agentId = agentId;
|
||||
(PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentSpkiSha256, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, ct);
|
||||
(PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentPublicKey, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, ct);
|
||||
_info = info;
|
||||
_agentKey = agentPublicKey;
|
||||
_udp = new UdpClient(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
_agentUdp = new IPEndPoint(IPAddress.Parse(host), udpPort);
|
||||
_rpc = new RpcSession<ProbeSimulator>(transport, _router, this, time ?? TimeProvider.System);
|
||||
@@ -152,6 +154,7 @@ public sealed class ProbeSimulator : IAsyncDisposable
|
||||
using var client = new System.Net.Sockets.TcpClient();
|
||||
await client.ConnectAsync(IPAddress.Loopback, tcpPort, ct);
|
||||
Stream stream = client.GetStream();
|
||||
await stream.WriteAsync(Knock.Build(Knock.DeriveKey(_identity, _agentKey!), TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds()), ct); // 预握手敲门
|
||||
if (tls)
|
||||
{
|
||||
#pragma warning disable CA5359 // 测试:Agent 的证书是自签名的,这里只验证数据面
|
||||
|
||||
+1
-1
@@ -50,7 +50,7 @@ probe.json:{ fleet: { server, identity_dir }, restart_at?, limits? } ← lim
|
||||
|
||||
Probe 启动
|
||||
├─ 连接 server:Fleet 长连接(用 probe_id 和私钥认证),fleet.hello 带上系统、架构和构建信息
|
||||
├─ 面板推送期望状态 probe_agents:{ version, agents: [ {agent_id, host, tcp_port, udp_ports, spki}... ] }
|
||||
├─ 面板推送期望状态 probe_agents:{ version, agents: [ {agent_id, host, tcp_port, udp_ports, public_key}... ] }
|
||||
├─ 调和:新增的 Agent → 建立点对点会话(peer 用途签名);被移除或禁用的 → 断开
|
||||
└─ 之后列表的任何变化(Agent 注册、禁用、地址或证书变化),面板都重新推送同一种期望状态;
|
||||
Probe 用 Fleet.Hosting.ConfigStore 把最新列表写回 agents.json(这是配置,不是测量数据;面板暂时连不上时,Probe 靠它继续测量)
|
||||
|
||||
+2
-2
@@ -12,13 +12,13 @@ MoShiro 的两条控制链路都建立在 Fleet 之上:
|
||||
| MoShiro 需要的能力 | 使用的 Fleet 机制 | MoShiro 提供的内容 |
|
||||
| --- | --- | --- |
|
||||
| 下发任务 | 期望状态 `fleet.state.sync` | 文档类型 `TaskSet`,包括 `tasks[]`、`peers`(允许连这个 Agent 的 Probe 公钥,已按线路开关过滤) |
|
||||
| 下发 Agent 列表给 Probe | 期望状态 `fleet.state.sync`(Probe 直连面板) | 文档类型 `probe_agents`:`version`、`agents[]`(`agent_id`、`host`、`tcp_port`、`udp_ports`、`spki`),已按线路开关过滤;Agent 集合或线路开关变化时重新推送 |
|
||||
| 下发 Agent 列表给 Probe | 期望状态 `fleet.state.sync`(Probe 直连面板) | 文档类型 `probe_agents`:`version`、`agents[]`(`agent_id`、`host`、`tcp_port`、`udp_ports`、`public_key`),`public_key` 是 Agent 面向 Probe 的 TLS 证书公钥(SPKI,base64):Probe 用它固定证书并派生敲门密钥;没有上报公钥的 Agent 不会下发。已按线路开关过滤;Agent 集合或线路开关变化时重新推送 |
|
||||
| 定时重启 | 通知 `moshiro.schedule`(B→A、B→P) | `{restart_at}`,形如 `04:30+08:00`,空表示关闭;连上时补发、管理端改了设置就群发。节点本地写了 `restart_at` 或 `MOSHIRO_RESTART_AT` 时以本地为准;节点只是到点自己有序退出,由服务管理器拉起 |
|
||||
| 更新 | 通知 `moshiro.update.offer` / `moshiro.update.apply`(B→A、B→P),心跳附加字段 `update` | `offer` 带 CI 签名的发布清单原文和签名,节点只写进数据目录;`apply` 请求更新;`update` 是宿主机更新器的状态(`state`、`current`、`available`、`error`、`at`),见 [deployment.md](deployment.md) |
|
||||
| 上报原始块 | 发件箱,二进制通道 **16** | `RawChunk`(自描述,见 [measurement.md](measurement.md) 第 4 节),`msg_id = chunk_id` |
|
||||
| Probe 交付列块 | 点对点会话的发件箱,二进制通道 **17** | 列块:`u32 stream_id \| u32 first_seq \| u32 count \| 列数据(Brotli)` |
|
||||
| 节点自身状态 | `fleet.heartbeat` 的附加字段 | Agent:`sessions`、`probes[]`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes`、`update`;Probe:只有 `update` |
|
||||
| 版本与能力 | `fleet.hello` | capabilities 包括 `moshiro.kinds:<列表>`、`moshiro.blocks:<列表>`;Agent 附加 `platform`(系统、架构、版本、tag、提交、构建时间)、`tls_spki`、`probe_ports`、`local_limits` 摘要;Probe 附加 `platform` |
|
||||
| 版本与能力 | `fleet.hello` | capabilities 包括 `moshiro.kinds:<列表>`、`moshiro.blocks:<列表>`;Agent 附加 `platform`(系统、架构、版本、tag、提交、构建时间)、`tls_pubkey`(面向 Probe 的证书公钥,面板自己算指纹)、`addresses`(本机公网地址)、`probe_ports`、`local_limits` 摘要;Probe 附加 `platform` |
|
||||
|
||||
## 2. `moshiro.*` 方法
|
||||
|
||||
|
||||
+3
-2
@@ -47,7 +47,7 @@ Agent、Backend、Probe 之间互相不信任。这份文档只回答三个问
|
||||
| G1 | 节点不验证 Backend 的身份,只靠 TLS 加公共 CA | `Fleet.Link` 里没有证书固定,也没有应用层验证 | 链路上任一环节被攻破,就能给节点下发任务集、Agent 名单、对端公钥、重启计划(更新清单有签名,不受影响) |
|
||||
| G2 | Probe 的私钥由 Backend 生成 | `ProbeService` 里的 `NodeIdentity.Generate`(创建和轮换) | Backend 接触过 Probe 的私钥,可以冒充它;Agent 的密钥是自己生成的,没有这个问题 |
|
||||
| G3 | Probe 会拨 Backend 给的任意 `host:port` | `AgentDirectory`、`AgentConnection` 不做任何地址检查;本地 `limits` 只有速率和 pps | 被攻破的 Backend 能让客户内网里的 Probe 去探测内网。固定 SPKI 只能防止建立会话,防不了探测本身 |
|
||||
| G4 | Agent 的 TCP 端口对任何人完成 TLS 握手并出示证书 | `TcpAcceptor`、`PeerListener` 在 `AcceptTcpClient` 之后直接 `AuthenticateAsServer` | 自签证书、TLS 参数、JARM 都能被扫描器指纹识别,一眼看出这是个服务 |
|
||||
| G4(已修,阶段二) | Agent 的 TCP 端口对任何人完成 TLS 握手并出示证书 | `TcpAcceptor`、`PeerListener` 在 `AcceptTcpClient` 之后直接 `AuthenticateAsServer` | 自签证书、TLS 参数、JARM 都能被扫描器指纹识别,一眼看出这是个服务 |
|
||||
| G5 | Backend 入库只限制了单块的大小 | `MaxDecodedChunkBytes` 等;没有找到按节点的速率或配额 | 恶意节点可以灌满磁盘 |
|
||||
| G6 | 对端公钥、Agent 地址全由 Backend 决定 | `TaskSet.Peers`、`ProbeAgentList` | 被攻破的 Backend 能把恶意 Probe 加进 Agent 的白名单,或把 Probe 指向恶意 Agent |
|
||||
| G7(已解决) | 方法的方向和角色没有统一声明:原来靠「谁注册了处理器」隐式决定 | `NodeConnection.HandleMessageAsync` 只按方法名分发到路由器;文档里的 `B→A`、`A→B` 只存在于 [protocol.md](protocol.md),没有对账 | 新增一个方法时,方向写反、角色写错(例如 Probe 调了只给 Agent 的方法)不会被任何东西发现 |
|
||||
@@ -226,6 +226,7 @@ Agent 的 `max_total_mbps` 默认是 1200,对 1 核 1 GB 的小机器没有意
|
||||
- `K_pair = HKDF(ECDH(Probe 身份私钥, Agent 身份公钥), "moshiro-knock-v1")`:只有名册里的成员算得出;ECDH 和签名共用同一把 P-256 密钥时,用不同的 HKDF 标签做域隔离。
|
||||
- 时间窗 30 秒,接受前后各一个窗(容忍时钟偏差);`nonce` 在窗内去重,防重放;每个来源地址限速,避免陌生人用随机数据消耗 HMAC 计算。
|
||||
- 这是破坏性变更:旧 Probe 不发敲门包,会被拒绝。
|
||||
- **已实现**:`Fleet.Identity.Knock`(构造与核对)和 `KnockGate`(读 32 字节、名册、nonce 去重、按来源限速,默认每来源每 10 秒 300 次);控制端口(`PeerListener`)和数据端口(`TcpAcceptor`)共用同一个关卡,Probe 一侧 `PeerConnector` 和 `TcpLink` 在 TCP 连接建立后、TLS 之前发敲门包。名册是期望状态里的 `peers`(Probe 的对等面公钥)。Probe 需要 Agent 的证书公钥来派生配对密钥,所以 `probe_agents` 下发的是 `public_key`(Agent 在 hello 里上报,面板自己算指纹)。陌生连接得到的是被关闭的连接(Windows 上可能表现为重置),没有任何字节。
|
||||
|
||||
**局限,必须说清楚**:TCP 的 SYN-ACK 由内核发出,用户态进程拦不住,所以端口在扫描器眼里仍是「开着,但不说话」,**不是「没监听」**。要做到真正的没监听,有两条路:
|
||||
|
||||
@@ -264,7 +265,7 @@ Agent 和 Probe 的核心永远是纯采集器。**所有敏感能力**(exec
|
||||
```text
|
||||
零 协议方向声明表 + 收发校验 + 对账测试 G7,第 8 节;同时把 NodeConnection 按职责拆开(第 13.2 节) 已完成,不破坏
|
||||
一 Probe 自己生成密钥 + 加入令牌注册 + 密钥按用途拆开 G2、G11 已完成,破坏性:旧 probe.json 失效
|
||||
二 预握手敲门(等级一) G4;先基于现有的对端公钥列表,名册出来后换成名册 破坏性:旧 Probe 被拒绝
|
||||
二 预握手敲门(等级一) G4;先基于现有的对端公钥列表,名册出来后换成名册 已完成,破坏性:旧 Probe 被拒绝
|
||||
三 操作员凭证 + 名册 + 节点验签与持久化 G1 的大部分、G3、G6;webauthn 和 ecdsa 先做,oidc 随后 破坏性:节点要固定操作员公钥并重新注册
|
||||
四 测量数据的可信 + 入库配额和输入校验 + CSP G5、G8、G9、G10、T5(第 9.1、9.2 节) 破坏性:块格式加签名
|
||||
五 Backend 应用层身份固定 + 下行消息签名 G1 的剩余部分(链路劫持) 破坏性
|
||||
|
||||
+2
-1
@@ -37,7 +37,8 @@ CREATE TABLE agents (
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
status TEXT NOT NULL, -- pending | active | disabled
|
||||
pubkey BLOB NOT NULL, -- ECDSA P-256 身份公钥
|
||||
tls_spki BLOB NOT NULL, -- 面向 Probe 的证书指纹
|
||||
tls_spki BLOB NOT NULL, -- 面向 Probe 的证书指纹(由 tls_pubkey 算出)
|
||||
tls_pubkey BLOB, -- 面向 Probe 的证书公钥,Agent 在 hello 里上报;Probe 用它固定证书、派生敲门密钥
|
||||
public_host TEXT NOT NULL, -- 自动识别(Agent 上报的公网地址),手填后标记 public_host_manual,不再被覆盖
|
||||
probe_tcp_port INTEGER NOT NULL, probe_udp_ports TEXT NOT NULL,
|
||||
labels TEXT NOT NULL DEFAULT '[]',
|
||||
|
||||
Reference in new issue
Block a user