fix(nodes): locate agents behind a proxy and retry missing countries [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful

This commit is contained in:
justme committed 2026-10-18 00:06:39 +08:00
1 parent 1145a53004
commit 0d532caf08
13 files changed
+93 -15

No files matched your search

@@ -6,6 +6,17 @@ namespace Fleet.Server;
public sealed record EnrollmentContext(string? RemoteIp, JsonElement? ProductMeta); public sealed record EnrollmentContext(string? RemoteIp, JsonElement? ProductMeta);
/// <summary>
/// 节点的真实来源地址:前面有反向代理时,套接字上只有代理的地址,产品在这里决定信不信转发头。
/// 没有注册就用套接字地址。
/// </summary>
public interface IClientAddressSource
{
/// <param name="socket">套接字上看到的对端地址。</param>
/// <param name="header">按名字取请求头的所有值。</param>
string? Resolve(System.Net.IPAddress? socket, Func<string, IEnumerable<string>> header);
}
/// <summary> /// <summary>
/// 服务器的身份密钥:操作员用 passkey 把它的公钥签进操作员集合。有它的时候,认证挑战会带上服务器的签名,节点据此确认对面是真的服务器, /// 服务器的身份密钥:操作员用 passkey 把它的公钥签进操作员集合。有它的时候,认证挑战会带上服务器的签名,节点据此确认对面是真的服务器,
/// 并在这条链路上用应用层加密。没有注册就是旧的行为(没有服务器身份)。 /// 并在这条链路上用应用层加密。没有注册就是旧的行为(没有服务器身份)。
@@ -38,7 +49,8 @@ public enum NodeEventKind
Heartbeat, Heartbeat,
} }
public sealed record NodeEvent(NodeEventKind Kind, NodeRecord Node, long AtMs, HelloParams? Hello = null, JsonElement? Extra = null, string? Reason = null); /// <param name="RemoteIp">这条连接的来源地址(经过 <see cref="IClientAddressSource"/>);没有就是 null。</param>
public sealed record NodeEvent(NodeEventKind Kind, NodeRecord Node, long AtMs, HelloParams? Hello = null, JsonElement? Extra = null, string? Reason = null, string? RemoteIp = null);
public interface INodeEventSink public interface INodeEventSink
{ {
@@ -73,7 +73,7 @@ public static class FleetEnrollmentEndpoints
return Problem(StatusCodes.Status400BadRequest, FleetErrors.InvalidParams, "request body is not valid"); return Problem(StatusCodes.Status400BadRequest, FleetErrors.InvalidParams, "request body is not valid");
} }
Result<EnrollResponse> result = await service.EnrollAsync(request, http.Connection.RemoteIpAddress?.ToString(), http.RequestAborted); Result<EnrollResponse> result = await service.EnrollAsync(request, ClientIp(http), http.RequestAborted);
if (result.TryGet(out EnrollResponse? response, out FleetError? error)) if (result.TryGet(out EnrollResponse? response, out FleetError? error))
{ {
return Results.Json(response, EnrollmentJsonContext.Default.EnrollResponse, statusCode: StatusCodes.Status201Created); return Results.Json(response, EnrollmentJsonContext.Default.EnrollResponse, statusCode: StatusCodes.Status201Created);
@@ -88,6 +88,11 @@ public static class FleetEnrollmentEndpoints
return Problem(status, error.Code, error.Message); return Problem(status, error.Code, error.Message);
} }
private static string? ClientIp(HttpContext http) =>
http.RequestServices.GetService<IClientAddressSource>() is { } source
? source.Resolve(http.Connection.RemoteIpAddress, name => http.Request.Headers[name].Select(v => v ?? ""))
: http.Connection.RemoteIpAddress?.ToString();
private static async Task<EnrollRequest?> ReadBodyAsync(HttpContext http, CancellationToken ct) private static async Task<EnrollRequest?> ReadBodyAsync(HttpContext http, CancellationToken ct)
{ {
if (http.Request.ContentLength > MaxBodyBytes) if (http.Request.ContentLength > MaxBodyBytes)
@@ -67,9 +67,9 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier
_connections.TryGetValue(node, out NodeConnection? c) ? c.Session : null; _connections.TryGetValue(node, out NodeConnection? c) ? c.Session : null;
/// <summary>处理一条 WebSocket 连接,直到连接结束。</summary> /// <summary>处理一条 WebSocket 连接,直到连接结束。</summary>
public async Task RunConnectionAsync(WebSocket socket, CancellationToken ct) public async Task RunConnectionAsync(WebSocket socket, string? remoteIp, CancellationToken ct)
{ {
using var connection = new NodeConnection(this, socket); using var connection = new NodeConnection(this, socket, remoteIp);
await connection.RunAsync(ct); await connection.RunAsync(ct);
} }
@@ -128,7 +128,10 @@ public static class FleetServerExtensions
} }
using System.Net.WebSockets.WebSocket socket = await http.WebSockets.AcceptWebSocketAsync(); using System.Net.WebSockets.WebSocket socket = await http.WebSockets.AcceptWebSocketAsync();
await http.RequestServices.GetRequiredService<FleetLinkHost>().RunConnectionAsync(socket, http.RequestAborted); string? remote = http.RequestServices.GetService<IClientAddressSource>() is { } source
? source.Resolve(http.Connection.RemoteIpAddress, name => http.Request.Headers[name].Select(v => v ?? ""))
: http.Connection.RemoteIpAddress?.ToString();
await http.RequestServices.GetRequiredService<FleetLinkHost>().RunConnectionAsync(socket, remote, http.RequestAborted);
}); });
return app; return app;
} }
@@ -102,7 +102,7 @@ internal sealed partial class NodeConnection
private async Task FireAsync(NodeEventKind kind, CancellationToken ct, HelloParams? hello = null, JsonElement? extra = null) private async Task FireAsync(NodeEventKind kind, CancellationToken ct, HelloParams? hello = null, JsonElement? extra = null)
{ {
var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra, kind == NodeEventKind.Offline ? _closeReason : null); var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra, kind == NodeEventKind.Offline ? _closeReason : null, _remoteIp);
foreach (INodeEventSink sink in _host.Sinks) foreach (INodeEventSink sink in _host.Sinks)
{ {
try try
@@ -34,10 +34,13 @@ internal sealed partial class NodeConnection : IDisposable
private bool _helloDone; private bool _helloDone;
private int _violations; private int _violations;
public NodeConnection(FleetLinkHost host, WebSocket socket) private readonly string? _remoteIp;
public NodeConnection(FleetLinkHost host, WebSocket socket, string? remoteIp = null)
{ {
_host = host; _host = host;
_socket = socket; _socket = socket;
_remoteIp = remoteIp;
} }
public NodeId NodeId => _node!.Id; public NodeId NodeId => _node!.Id;
@@ -20,6 +20,7 @@ public static class ApiModule
services.AddSingleton<SseLimiter>(); services.AddSingleton<SseLimiter>();
services.AddSingleton<MoShiro.Backend.Application.IReleaseKey, EmbeddedReleaseKey>(); services.AddSingleton<MoShiro.Backend.Application.IReleaseKey, EmbeddedReleaseKey>();
services.AddSingleton<ClientAddressResolver>(); services.AddSingleton<ClientAddressResolver>();
services.AddSingleton<Fleet.Server.IClientAddressSource>(p => p.GetRequiredService<ClientAddressResolver>());
services.Configure<JsonOptions>(options => services.Configure<JsonOptions>(options =>
{ {
JsonSerializerOptions json = options.SerializerOptions; JsonSerializerOptions json = options.SerializerOptions;
@@ -11,7 +11,7 @@ namespace MoShiro.Backend.Api;
/// 且从右往左跳过所有可信代理,取第一个不可信的地址(客户端自己伪造的在最左边,不会被采信)。 /// 且从右往左跳过所有可信代理,取第一个不可信的地址(客户端自己伪造的在最左边,不会被采信)。
/// 没配置就一律用套接字地址:宁可限流和审计里是代理的地址,也不让客户端自己报地址。 /// 没配置就一律用套接字地址:宁可限流和审计里是代理的地址,也不让客户端自己报地址。
/// </summary> /// </summary>
public sealed class ClientAddressResolver public sealed class ClientAddressResolver : Fleet.Server.IClientAddressSource
{ {
private readonly IPNetwork[] _trusted; private readonly IPNetwork[] _trusted;
private readonly string _header; private readonly string _header;
@@ -43,9 +43,13 @@ public sealed class ClientAddressResolver
return _trusted.Any(n => n.Contains(normalized)); return _trusted.Any(n => n.Contains(normalized));
} }
public string Resolve(HttpContext http) string? Fleet.Server.IClientAddressSource.Resolve(IPAddress? socket, Func<string, IEnumerable<string>> header) => Resolve(socket, header);
public string Resolve(HttpContext http) =>
Resolve(http.Connection.RemoteIpAddress, name => http.Request.Headers[name].Select(v => v ?? ""));
private string Resolve(IPAddress? remote, Func<string, IEnumerable<string>> headers)
{ {
IPAddress? remote = http.Connection.RemoteIpAddress;
if (remote is null) if (remote is null)
{ {
return "unknown"; return "unknown";
@@ -56,7 +60,7 @@ public sealed class ClientAddressResolver
return remote.ToString(); return remote.ToString();
} }
string[] chain = [.. http.Request.Headers[_header].SelectMany(v => (v ?? "").Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))]; string[] chain = [.. headers(_header).SelectMany(v => (v ?? "").Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))];
for (int i = chain.Length - 1; i >= 0; i--) for (int i = chain.Length - 1; i >= 0; i--)
{ {
if (!IPAddress.TryParse(chain[i], out IPAddress? hop)) if (!IPAddress.TryParse(chain[i], out IPAddress? hop))
@@ -115,10 +115,15 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
await _events.AddAsync(id, "agent_offline", nodeEvent.AtMs, null, ct); await _events.AddAsync(id, "agent_offline", nodeEvent.AtMs, null, ct);
break; break;
case NodeEventKind.Hello when nodeEvent.Hello is { } hello: case NodeEventKind.Hello when nodeEvent.Hello is { } hello:
await ApplyHelloAsync(id, hello, ct); await ApplyHelloAsync(id, hello, nodeEvent.RemoteIp, ct);
break; break;
case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra: case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra:
await _nodes.SetVersionAsync(id.Value, NodeKinds.Agent, nodeEvent.Node.NodeVersion, ct); await _nodes.SetVersionAsync(id.Value, NodeKinds.Agent, nodeEvent.Node.NodeVersion, ct);
if (_nodes.ShouldRetryGeo(id.Value) && await _agents.GetAsync(id, ct) is { } unlocated)
{
_nodes.ObserveAgentInBackground(id.Value, GeoHostOf(unlocated, nodeEvent.RemoteIp), null);
}
try try
{ {
AgentRuntimeState state = _runtime.Get(id); AgentRuntimeState state = _runtime.Get(id);
@@ -145,6 +150,12 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
} }
} }
/// <summary>查国家和运营商用哪个地址:对外地址优先,其次 IPv6 地址,都没有(NAT 后面、或者没有上报公网地址)才用这条连接的来源地址。</summary>
private static string? GeoHostOf(AgentRecord? agent, string? remoteIp) =>
!string.IsNullOrEmpty(agent?.PublicHost) ? agent.PublicHost
: !string.IsNullOrEmpty(agent?.Ipv6Host) ? agent.Ipv6Host
: remoteIp;
/// <summary>Probe 自己连上来了:hello 里带着它的系统、架构和构建信息;上线下线只推给 SSE,不记事件。</summary> /// <summary>Probe 自己连上来了:hello 里带着它的系统、架构和构建信息;上线下线只推给 SSE,不记事件。</summary>
private async ValueTask OnProbeEventAsync(NodeId id, NodeEvent nodeEvent, CancellationToken ct) private async ValueTask OnProbeEventAsync(NodeId id, NodeEvent nodeEvent, CancellationToken ct)
{ {
@@ -213,7 +224,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
} }
} }
private async ValueTask ApplyHelloAsync(NodeId id, HelloParams hello, CancellationToken ct) private async ValueTask ApplyHelloAsync(NodeId id, HelloParams hello, string? remoteIp, CancellationToken ct)
{ {
var kinds = new List<string>(); var kinds = new List<string>();
foreach (string capability in hello.Capabilities) foreach (string capability in hello.Capabilities)
@@ -278,7 +289,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
// 端口、证书指纹或对外地址变了,Probe 需要新的列表 // 端口、证书指纹或对外地址变了,Probe 需要新的列表
AgentRecord? after = await _agents.GetAsync(id, ct); AgentRecord? after = await _agents.GetAsync(id, ct);
_nodes.ObserveAgentInBackground(id.Value, after?.PublicHost, meta.Platform); _nodes.ObserveAgentInBackground(id.Value, GeoHostOf(after, remoteIp), meta.Platform);
await _releases.OfferToNodeAsync(id, ct); await _releases.OfferToNodeAsync(id, ct);
await _schedule.SendToNodeAsync(id, ct); await _schedule.SendToNodeAsync(id, ct);
if (before is not null && after is not null if (before is not null && after is not null
@@ -26,6 +26,7 @@ public sealed class NodeInfoService
private readonly TimeProvider _time; private readonly TimeProvider _time;
private readonly ConcurrentDictionary<string, NodeMeta> _cache = new(); private readonly ConcurrentDictionary<string, NodeMeta> _cache = new();
private readonly ConcurrentDictionary<string, byte> _inflight = new(); private readonly ConcurrentDictionary<string, byte> _inflight = new();
private readonly ConcurrentDictionary<string, long> _lastTry = new();
private Task? _load; private Task? _load;
public NodeInfoService(INodeMetaStore store, IGeoLookup geo, TimeProvider time) public NodeInfoService(INodeMetaStore store, IGeoLookup geo, TimeProvider time)
@@ -119,6 +120,7 @@ public sealed class NodeInfoService
public async ValueTask ObserveAddressAsync(string nodeId, string kind, string? host, CancellationToken ct) public async ValueTask ObserveAddressAsync(string nodeId, string kind, string? host, CancellationToken ct)
{ {
host = host?.Trim(); host = host?.Trim();
_lastTry[nodeId] = Now();
if (string.IsNullOrEmpty(host) || !IsPublic(host) || !_inflight.TryAdd(nodeId, 0)) if (string.IsNullOrEmpty(host) || !IsPublic(host) || !_inflight.TryAdd(nodeId, 0))
{ {
return; return;
@@ -162,6 +164,17 @@ public sealed class NodeInfoService
} }
} }
/// <summary>国家还没查到、并且离上一次尝试已经过了重试间隔:调用方可以再查一次(心跳里调,不会每次都真去查)。</summary>
public bool ShouldRetryGeo(string nodeId)
{
if (_cache.TryGetValue(nodeId, out NodeMeta? meta) && (meta.Info.Country is not null || meta.Info.IspManual))
{
return false;
}
return !_lastTry.TryGetValue(nodeId, out long at) || Now() - at >= s_retryAfter.TotalMilliseconds;
}
/// <summary>Agent 报告某个 Probe 连着它:在后台用 Probe 的来源地址查国家和运营商(系统信息是 Probe 自己连上面板时报的)。出错只吞掉。</summary> /// <summary>Agent 报告某个 Probe 连着它:在后台用 Probe 的来源地址查国家和运营商(系统信息是 Probe 自己连上面板时报的)。出错只吞掉。</summary>
public void ObserveProbeInBackground(string probeId, string? remote) => public void ObserveProbeInBackground(string probeId, string? remote) =>
_ = Task.Run(async () => _ = Task.Run(async () =>
@@ -48,4 +48,12 @@ public sealed class ClientAddressTests
[Fact] [Fact]
public void AMalformedTrustedProxyEntry_FailsAtStartup() => public void AMalformedTrustedProxyEntry_FailsAtStartup() =>
Assert.Throws<InvalidOperationException>(() => new ClientAddressResolver(new BackendOptions { TrustedProxies = "banana" })); Assert.Throws<InvalidOperationException>(() => new ClientAddressResolver(new BackendOptions { TrustedProxies = "banana" }));
[Fact]
public void TheFleetLinkAsksTheSameResolver_SoANodeBehindAProxyKeepsItsRealAddress()
{
Fleet.Server.IClientAddressSource source = new ClientAddressResolver(new BackendOptions { TrustedProxies = "127.0.0.1" });
Assert.Equal("198.51.100.7", source.Resolve(IPAddress.Parse("127.0.0.1"), _ => ["198.51.100.7"]));
Assert.Equal("203.0.113.9", source.Resolve(IPAddress.Parse("203.0.113.9"), _ => ["198.51.100.7"]));
}
} }
@@ -102,6 +102,24 @@ public sealed partial class NodeInfoTests
Assert.Null(Assert.Single(await service.ListAsync(CancellationToken.None)).Country); Assert.Null(Assert.Single(await service.ListAsync(CancellationToken.None)).Country);
} }
[Fact]
public async Task Service_AnUnlocatedNode_IsRetriedFromTheHeartbeat_UntilItHasACountry()
{
var time = new FakeTimeProvider(DateTimeOffset.Parse("2026-10-06T00:00:00Z"));
var service = new NodeInfoService(new MemoryStore(), new FakeGeo(null), time);
Assert.True(service.ShouldRetryGeo("agt_1"));
await service.ObserveAddressAsync("agt_1", NodeKinds.Agent, "", CancellationToken.None);
Assert.False(service.ShouldRetryGeo("agt_1"));
time.Advance(TimeSpan.FromMinutes(11));
Assert.True(service.ShouldRetryGeo("agt_1"));
var found = new NodeInfoService(new MemoryStore(), new FakeGeo(new GeoInfo("jp", 1, "Example", false)), time);
await found.ObserveAddressAsync("agt_2", NodeKinds.Agent, "8.8.8.8", CancellationToken.None);
time.Advance(TimeSpan.FromMinutes(11));
Assert.False(found.ShouldRetryGeo("agt_2"));
}
[Fact] [Fact]
public async Task Service_PlatformIsSavedAndSurvivesARestart() public async Task Service_PlatformIsSavedAndSurvivesARestart()
{ {
+1 -1
View File
@@ -199,7 +199,7 @@ OpenWrt 默认不带 OpenSSL:`opkg install libopenssl3`(较新的快照和 2
没有公网入站的家宽机器上跑 Backend 时,用隧道(Cloudflare Tunnel 之类)把一个公网域名转到 Backend 的端口;Backend 只写一个监听地址(同时写 `0.0.0.0` 和 `[::]` 会报地址已被占用);Probe 的 `fleet.server` 就是这个面板地址。小 VPS 上把 `limits.max_total_mbps` 调低。家宽的 IPv6 前缀会在路由器重新拨号(比如停电)后变化,用 IPv6 直连时地址要跟着换;走隧道的域名不受影响。 没有公网入站的家宽机器上跑 Backend 时,用隧道(Cloudflare Tunnel 之类)把一个公网域名转到 Backend 的端口;Backend 只写一个监听地址(同时写 `0.0.0.0` 和 `[::]` 会报地址已被占用);Probe 的 `fleet.server` 就是这个面板地址。小 VPS 上把 `limits.max_total_mbps` 调低。家宽的 IPv6 前缀会在路由器重新拨号(比如停电)后变化,用 IPv6 直连时地址要跟着换;走隧道的域名不受影响。
**国家和运营商**:Backend 用 Probe、Agent 的来源地址向 IP 归属服务查一次,展示在国旗和运营商图标上(`MoShiro__GeoLookupUrl`,缺省 `https://ip.007995.xyz`;留空就不查;这个服务只供公益使用,署名和使用条款由部署的人自己遵守)。Probe、Agent 自己不访问外部服务,所以客户的防火墙审计里不会多出出站连接;Backend 发出去的只有公网 IP(内网地址不发)。 **国家和运营商**:Backend 用 Probe、Agent 的来源地址向 IP 归属服务查一次,展示在国旗和运营商图标上(`MoShiro__GeoLookupUrl`,缺省 `https://ip.007995.xyz`;留空就不查;这个服务只供公益使用,署名和使用条款由部署的人自己遵守)。Probe、Agent 自己不访问外部服务,所以客户的防火墙审计里不会多出出站连接;Backend 发出去的只有公网 IP(内网地址不发)。Agent 的地址优先用它的对外地址(`public_host`),没有就用 IPv6 地址,再没有(NAT 后面、没有上报公网地址)才用它连面板的来源地址;面板前面有反向代理时,要配置 `MoShiro__TrustedProxies`,否则来源地址是代理的内网地址,这类节点查不到国家。没查到的节点每隔十分钟在心跳里重试一次。
**从别处看这个面板**:本机跑前端,在页面里把数据源指向面板的地址,用 API 令牌访问,Backend 设 `MoShiro__CorsOrigins` 放行前端的来源(见 [web/README.md](../../web/README.md))。 **从别处看这个面板**:本机跑前端,在页面里把数据源指向面板的地址,用 API 令牌访问,Backend 设 `MoShiro__CorsOrigins` 放行前端的来源(见 [web/README.md](../../web/README.md))。