diff --git a/README.md b/README.md index 517f88d..21bd5b7 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,7 @@ In the application, generate or copy the monitoring SSH key, add the Hub profile The current development branch implements Windows SSH monitoring, the Hub/Node WireGuard installer, directional Links, one-time enrollment, mTLS Agent and Operator identities, certificate revocation/re-enrollment, SQLite control state, audit, authenticated event streaming, Windows Control API integration, and a bounded durable Agent buffer with downsampling. -Still planned: reconnect reconciliation, kill-switch failure tests, a 50–100 Node load test, signed Windows installer, and desktop/mobile clients for additional platforms. +Reconnect reconciliation is implemented with a durable SQLite marker: after a Node returns, the Hub reapplies the latest effective disabled policies and clears the marker only after the firewall confirms success. Still planned: kill-switch integration tests, a 50–100 Node load test, signed Windows installer, and desktop/mobile clients for additional platforms. ## License and project policy diff --git a/docs/architecture.md b/docs/architecture.md index 9ea0a74..4211451 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -74,7 +74,7 @@ Active -> Expired -> Disabled Пустая политика не означает `allow all`. Ответный трафик существующего соединения разрешается stateful-правилом, обратное новое соединение требует отдельного Link. -При `Disconnect` Hub сначала блокирует направление в nftables, затем фиксирует `Disabled`. Если подтверждение узла недоступно, интерфейс показывает `Partial`, а обязательное состояние применяется после reconnect. +При `Disconnect` Hub сначала блокирует направление в nftables, затем фиксирует `Disabled`. Если подтверждение узла недоступно, интерфейс показывает `Partial`. После reconnect heartbeat устанавливает долговечный marker в SQLite, Hub повторно применяет последнюю эффективную отключённую политику и снимает marker только после успешного подтверждения firewall. При перезапуске Control Hub незавершённый marker сохраняется. ## 5. Сценарий AI-агента diff --git a/docs/i18n/README.de.md b/docs/i18n/README.de.md index 7a1880f..a48ed12 100644 --- a/docs/i18n/README.de.md +++ b/docs/i18n/README.de.md @@ -32,7 +32,7 @@ sudo ./ochenstarik-server-monitor-manager.sh hub Es gibt kein gemeinsames Root-Passwort; der private WireGuard-Schlüssel bleibt auf dem Node. Monitoring-, Terminal-, Agent-, Operator- und KI-Automationsidentitäten sind getrennt. SSH nutzt einen forced-command ohne Shell/PTY/Forwarding; mTLS begrenzt Rollen; nftables erlaubt nur explizite Links; SQLite speichert Sollzustand und Audit vor der Firewalländerung. -`v0.1.0-alpha.4` ist eine Testversion. Der aktuelle Entwicklungszweig enthält Windows-Client, Hub/Node-Installer, Links, mTLS, Zertifikatswiderruf und erneute Registrierung, SQLite, Audit, Events und einen begrenzten Offline-Puffer mit Downsampling. Offen sind Reconnect-Abgleich, Lasttest mit 50–100 Nodes und signierter Windows-Installer. +`v0.1.0-alpha.4` ist eine Testversion. Der aktuelle Entwicklungszweig enthält Windows-Client, Hub/Node-Installer, Links, mTLS, Zertifikatswiderruf und erneute Registrierung, SQLite, Audit, Events, einen begrenzten Offline-Puffer mit Downsampling und dauerhaften Reconnect-Abgleich. Offen sind der Lasttest mit 50–100 Nodes und ein signierter Windows-Installer. ## Lizenz diff --git a/docs/i18n/README.hi.md b/docs/i18n/README.hi.md index b053373..b8f245e 100644 --- a/docs/i18n/README.hi.md +++ b/docs/i18n/README.hi.md @@ -32,7 +32,7 @@ Hub पर UDP `51820` और TCP `7443` खोलें, Node codes बना Shared root password नहीं है और Node की WireGuard private key Node से बाहर नहीं जाती। Monitoring, terminal, Agent, Operator और AI automation identities अलग हैं। SSH forced-command shell/PTY/forwarding नहीं देता; mTLS roles सीमित करता है; nftables केवल explicit Links स्वीकारता है; SQLite पहले desired state और audit सहेजता है। -`v0.1.0-alpha.4` testing release है। Current development branch में Windows client, Hub/Node installer, Links, mTLS, certificate revocation और re-enrollment, SQLite, audit, event stream तथा downsampling वाला सीमित offline buffer तैयार हैं। Reconnect reconciliation, 50–100 Node load test और signed Windows installer अभी बाकी हैं। +`v0.1.0-alpha.4` testing release है। Current development branch में Windows client, Hub/Node installer, Links, mTLS, certificate revocation और re-enrollment, SQLite, audit, event stream, downsampling वाला सीमित offline buffer और durable reconnect reconciliation तैयार हैं। 50–100 Node load test और signed Windows installer अभी बाकी हैं। ## लाइसेंस diff --git a/docs/i18n/README.ru.md b/docs/i18n/README.ru.md index 6a7219d..4ebea6f 100644 --- a/docs/i18n/README.ru.md +++ b/docs/i18n/README.ru.md @@ -38,7 +38,7 @@ sudo ./ochenstarik-server-monitor-manager.sh hub Нет общих root-паролей; приватный WireGuard-ключ Node не покидает Node. Идентичности monitoring, terminal, Agent, Operator и AI-автоматизации разделены. Monitoring SSH использует forced-command без shell/PTY/forwarding. Agent может отправлять heartbeat только своего узла, Operator управляет inventory и Links. Отключение Link сначала сохраняется в SQLite, затем удаляет разрешение nftables; повтор запроса не повторяет побочный эффект. -`v0.1.0-alpha.4` предназначен для тестирования. В текущей ветке разработки уже готовы Windows SSH-monitoring, Hub/Node installer, Links, mTLS, отзыв и повторная регистрация сертификатов, SQLite, аудит, поток событий и ограниченный offline-буфер Agent с downsampling. Остались reconnect reconciliation, нагрузочный тест 50–100 Node и подписанный Windows installer. До стабильного релиза используйте тестовые или резервируемые серверы. +`v0.1.0-alpha.4` предназначен для тестирования. В текущей ветке разработки уже готовы Windows SSH-monitoring, Hub/Node installer, Links, mTLS, отзыв и повторная регистрация сертификатов, SQLite, аудит, поток событий, ограниченный offline-буфер Agent с downsampling и долговечный reconnect reconciliation. Остались нагрузочный тест 50–100 Node и подписанный Windows installer. До стабильного релиза используйте тестовые или резервируемые серверы. ## Лицензия diff --git a/docs/roadmap.md b/docs/roadmap.md index 40e1532..bf6e801 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -6,8 +6,8 @@ - [x] выбрать звёздную архитектуру Hub/Node для первого Mesh; - [x] разделить monitoring identity, terminal identity и AI-agent identity; - [x] описать направленные Links и kill switch; -- [ ] выбрать лицензию; -- [ ] добавить CI, форматирование, тесты и release checksum; +- [x] выбрать лицензию; +- [x] добавить CI, форматирование, тесты и release checksum; - [ ] объединить PR приложения и установщика в `main`. ## Этап 1 — Windows SSH MVP @@ -50,9 +50,9 @@ - [x] ручное добавление и удаление nftables ACL из Windows-клиента; - [x] политики по целевому `/32`, TCP/UDP и порту; - [x] TTL и автоматическое истечение; -- [ ] состояния Connecting, Active, Disconnecting, Partial, Disabled и Failed; +- [x] состояния Connecting, Active, Disconnecting, Partial, Disabled и Failed; - [x] версия политики и подтверждение применения на Hub; -- [ ] обязательное отключение после reconnect; +- [x] обязательное отключение после reconnect; - [x] локальный append-only JSONL-аудит операций Link; - [ ] интеграционные тесты kill switch и частичных отказов. diff --git a/docs/security-model.md b/docs/security-model.md index 5ff358a..1302bfb 100644 --- a/docs/security-model.md +++ b/docs/security-model.md @@ -58,7 +58,7 @@ Windows-клиент получает отдельный код `SMMDEV1`, по 1. немедленно удалить разрешающее правило на Hub; 2. записать желаемое состояние `Disabled` до отправки дополнительных команд; -3. запретить автоматическое восстановление после reboot/reconnect; +3. запретить автоматическое восстановление после reboot/reconnect долговечным SQLite-marker, который снимается только после подтверждения firewall; 4. получить подтверждение применённой версии политики; 5. показывать `Partial`, если подтверждение одного из узлов отсутствует; 6. сохранить обязательную операцию для временно недоступного Node. diff --git a/src/ServerMonitorManager.Control/ControlStore.cs b/src/ServerMonitorManager.Control/ControlStore.cs index 349cb47..0f72b51 100644 --- a/src/ServerMonitorManager.Control/ControlStore.cs +++ b/src/ServerMonitorManager.Control/ControlStore.cs @@ -62,6 +62,11 @@ public sealed class ControlStore(IOptions options) ); CREATE INDEX IF NOT EXISTS ix_metric_samples_node_time ON metric_samples(node_id, recorded_at DESC); + CREATE TABLE IF NOT EXISTS agent_reconciliation ( + node_id TEXT PRIMARY KEY REFERENCES agents(node_id) ON DELETE CASCADE, + required INTEGER NOT NULL, + requested_at TEXT NOT NULL + ); CREATE TABLE IF NOT EXISTS idempotency ( operation_key TEXT PRIMARY KEY, request_hash TEXT NOT NULL, @@ -341,7 +346,7 @@ public sealed class ControlStore(IOptions options) return Convert.ToInt32(await command.ExecuteScalarAsync(cancellationToken)) == 1; } - public async Task RecordHeartbeatAsync( + public async Task RecordHeartbeatAsync( AgentHeartbeat heartbeat, int nextHeartbeatSeconds, CancellationToken cancellationToken = default) @@ -357,11 +362,49 @@ public sealed class ControlStore(IOptions options) cancellationToken); if (cached is not null) { + var cachedReconciliationRequired = await IsAgentReconciliationRequiredAsync( + connection, transaction, heartbeat.NodeId, cancellationToken); await transaction.CommitAsync(cancellationToken); - return cached; + return new AgentHeartbeatMutation(cached, cachedReconciliationRequired); } var now = DateTimeOffset.UtcNow; + var previous = connection.CreateCommand(); + previous.Transaction = transaction; + previous.CommandText = "SELECT status, last_seen_at FROM agents WHERE node_id = $node;"; + previous.Parameters.AddWithValue("$node", heartbeat.NodeId); + string previousStatus; + DateTimeOffset? previousLastSeenAt; + await using (var reader = await previous.ExecuteReaderAsync(cancellationToken)) + { + if (!await reader.ReadAsync(cancellationToken)) + { + throw new InvalidOperationException("Unknown agent node id."); + } + previousStatus = reader.GetString(0); + previousLastSeenAt = reader.IsDBNull(1) ? null : DateTimeOffset.Parse(reader.GetString(1)); + } + var reconnectThreshold = TimeSpan.FromSeconds(Math.Max(60, nextHeartbeatSeconds * 3)); + var requiresReconciliation = previousStatus != "Online" + || previousLastSeenAt is null + || now - previousLastSeenAt >= reconnectThreshold; + if (requiresReconciliation) + { + var requireReconciliation = connection.CreateCommand(); + requireReconciliation.Transaction = transaction; + requireReconciliation.CommandText = """ + INSERT INTO agent_reconciliation(node_id, required, requested_at) + VALUES ($node, 1, $now) + ON CONFLICT(node_id) DO UPDATE SET + required = 1, + requested_at = excluded.requested_at; + """; + requireReconciliation.Parameters.AddWithValue("$node", heartbeat.NodeId); + requireReconciliation.Parameters.AddWithValue("$now", now.ToString("O")); + await requireReconciliation.ExecuteNonQueryAsync(cancellationToken); + } + var reconciliationRequired = await IsAgentReconciliationRequiredAsync( + connection, transaction, heartbeat.NodeId, cancellationToken); var update = connection.CreateCommand(); update.Transaction = transaction; update.CommandText = """ @@ -400,7 +443,22 @@ public sealed class ControlStore(IOptions options) SmmJsonContext.Default.AgentHeartbeatResponse, cancellationToken); await transaction.CommitAsync(cancellationToken); - return response; + return new AgentHeartbeatMutation(response, reconciliationRequired); + } + + public async Task CompleteAgentReconciliationAsync( + string nodeId, + CancellationToken cancellationToken = default) + { + await using var connection = await OpenAsync(cancellationToken); + var command = connection.CreateCommand(); + command.CommandText = """ + UPDATE agent_reconciliation + SET required = 0 + WHERE node_id = $node; + """; + command.Parameters.AddWithValue("$node", nodeId); + await command.ExecuteNonQueryAsync(cancellationToken); } public async Task> ListAgentsAsync(CancellationToken cancellationToken = default) @@ -733,6 +791,20 @@ public sealed class ControlStore(IOptions options) await transaction.RollbackAsync(cancellationToken); return null; } + if (existing.DesiredState == "Disabled" + || await HasNewerLinkAsync(connection, transaction, existing, cancellationToken)) + { + await WriteIdempotentAsync( + connection, + transaction, + $"link-disable:{actor}:{id}:{request.IdempotencyKey}", + Fingerprint(request, SmmJsonContext.Default.LinkPolicyDisableRequest), + existing, + SmmJsonContext.Default.LinkPolicy, + cancellationToken); + await transaction.CommitAsync(cancellationToken); + return new LinkMutation(existing, true); + } var now = DateTimeOffset.UtcNow; var link = existing with { @@ -822,6 +894,36 @@ public sealed class ControlStore(IOptions options) return result; } + public async Task> ListEffectiveLinksForNodeAsync( + string nodeId, + CancellationToken cancellationToken = default) + { + var result = new List(); + await using var connection = await OpenAsync(cancellationToken); + var command = connection.CreateCommand(); + command.CommandText = """ + SELECT current.* + FROM links AS current + WHERE (current.source_node_id = $node OR current.target_node_id = $node) + AND NOT EXISTS ( + SELECT 1 + FROM links AS newer + WHERE newer.source_node_id = current.source_node_id + AND newer.target_node_id = current.target_node_id + AND newer.protocol = current.protocol + AND newer.port = current.port + AND newer.version > current.version) + ORDER BY current.version; + """; + command.Parameters.AddWithValue("$node", nodeId); + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + result.Add(ReadLink(reader)); + } + return result; + } + private static void AddLinkParameters(SqliteCommand command, LinkPolicy link) { command.Parameters.AddWithValue("$id", link.Id); @@ -841,6 +943,23 @@ public sealed class ControlStore(IOptions options) command.Parameters.AddWithValue("$updated", link.UpdatedAt.ToString("O")); } + private static async Task IsAgentReconciliationRequiredAsync( + SqliteConnection connection, + SqliteTransaction transaction, + string nodeId, + CancellationToken cancellationToken) + { + var command = connection.CreateCommand(); + command.Transaction = transaction; + command.CommandText = """ + SELECT COALESCE(( + SELECT required FROM agent_reconciliation WHERE node_id = $node + ), 0); + """; + command.Parameters.AddWithValue("$node", nodeId); + return Convert.ToInt32(await command.ExecuteScalarAsync(cancellationToken)) == 1; + } + private static async Task ReadLinkAsync( SqliteConnection connection, SqliteTransaction transaction, @@ -855,6 +974,31 @@ public sealed class ControlStore(IOptions options) return await reader.ReadAsync(cancellationToken) ? ReadLink(reader) : null; } + private static async Task HasNewerLinkAsync( + SqliteConnection connection, + SqliteTransaction transaction, + LinkPolicy link, + CancellationToken cancellationToken) + { + var command = connection.CreateCommand(); + command.Transaction = transaction; + command.CommandText = """ + SELECT EXISTS( + SELECT 1 FROM links + WHERE source_node_id = $source + AND target_node_id = $target + AND protocol = $protocol + AND port = $port + AND version > $version); + """; + command.Parameters.AddWithValue("$source", link.SourceNodeId); + command.Parameters.AddWithValue("$target", link.TargetNodeId); + command.Parameters.AddWithValue("$protocol", link.Protocol); + command.Parameters.AddWithValue("$port", link.Port); + command.Parameters.AddWithValue("$version", link.Version); + return Convert.ToInt32(await command.ExecuteScalarAsync(cancellationToken)) == 1; + } + private static LinkPolicy ReadLink(SqliteDataReader reader) => new( reader.GetString(0), @@ -998,6 +1142,12 @@ public sealed record ControlIdentity(string Id, string Role); public sealed record LinkMutation(LinkPolicy Link, bool IsReplay); +public sealed record AgentHeartbeatMutation( + AgentHeartbeatResponse Response, + bool RequiresReconciliation); + +public sealed record LinkReconciliationResult(int Applied, int Failed); + public sealed record AgentReenrollmentMutation( CertificateReenrollmentTicket Ticket, IReadOnlyList Links, diff --git a/src/ServerMonitorManager.Control/LinkService.cs b/src/ServerMonitorManager.Control/LinkService.cs index d6aa972..a7393e3 100644 --- a/src/ServerMonitorManager.Control/LinkService.cs +++ b/src/ServerMonitorManager.Control/LinkService.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Text.Json; using ServerMonitorManager.Core; @@ -8,6 +9,8 @@ public sealed class LinkService( ILinkPolicyApplier applier, ControlEventBroker events) { + private readonly ConcurrentDictionary _reconciliationLocks = new(); + public async Task CreateAsync( LinkPolicyCreateRequest request, string actor, @@ -70,6 +73,54 @@ public sealed class LinkService( return link; } + public async Task ReconcileDisabledLinksForNodeAsync( + string nodeId, + CancellationToken cancellationToken) + { + var reconciled = 0; + var failed = 0; + var links = await store.ListEffectiveLinksForNodeAsync(nodeId, cancellationToken); + foreach (var candidate in links.Where(link => link.DesiredState == "Disabled")) + { + var gate = _reconciliationLocks.GetOrAdd(candidate.Id, static _ => new SemaphoreSlim(1, 1)); + await gate.WaitAsync(cancellationToken); + try + { + var current = (await store.ListEffectiveLinksForNodeAsync(nodeId, cancellationToken)) + .SingleOrDefault(link => link.Id == candidate.Id && link.DesiredState == "Disabled"); + if (current is null) + { + continue; + } + + var actor = $"system:reconnect:{nodeId}"; + var link = await store.SetLinkActualStateAsync( + current.Id, "Disconnecting", null, actor, cancellationToken) ?? current; + Publish("link.reconciling", link); + try + { + await applier.ApplyDisconnectAsync(link, cancellationToken); + link = await store.SetLinkActualStateAsync( + link.Id, "Disabled", null, actor, cancellationToken) ?? link; + Publish("link.disabled", link); + } + catch (Exception exception) when (exception is not OperationCanceledException) + { + link = await store.SetLinkActualStateAsync( + link.Id, "Partial", CompactError(exception), actor, cancellationToken) ?? link; + Publish("link.partial", link); + failed++; + } + reconciled++; + } + finally + { + gate.Release(); + } + } + return new LinkReconciliationResult(reconciled, failed); + } + private void Publish(string type, LinkPolicy link) => events.Publish( type, diff --git a/src/ServerMonitorManager.Control/Program.cs b/src/ServerMonitorManager.Control/Program.cs index 7487cc1..c52e469 100644 --- a/src/ServerMonitorManager.Control/Program.cs +++ b/src/ServerMonitorManager.Control/Program.cs @@ -221,6 +221,7 @@ agents.MapPost("/heartbeat", async ( AgentHeartbeat heartbeat, HttpContext context, ControlStore controlStore, + LinkService linkService, IOptions options, CancellationToken cancellationToken) => { @@ -251,16 +252,26 @@ agents.MapPost("/heartbeat", async ( try { - var response = await controlStore.RecordHeartbeatAsync( + var mutation = await controlStore.RecordHeartbeatAsync( heartbeat, options.Value.HeartbeatSeconds, cancellationToken); + if (mutation.RequiresReconciliation) + { + var reconciliation = await linkService.ReconcileDisabledLinksForNodeAsync( + heartbeat.NodeId, cancellationToken); + if (reconciliation.Failed == 0) + { + await controlStore.CompleteAgentReconciliationAsync( + heartbeat.NodeId, cancellationToken); + } + } var broker = context.RequestServices.GetRequiredService(); broker.Publish( "agent.heartbeat", heartbeat.NodeId, JsonSerializer.Serialize(heartbeat, SmmJsonContext.Default.AgentHeartbeat)); - return Results.Ok(response); + return Results.Ok(mutation.Response); } catch (IdempotencyConflictException) { diff --git a/tests/ServerMonitorManager.Control.Tests/ControlStoreTests.cs b/tests/ServerMonitorManager.Control.Tests/ControlStoreTests.cs index 13cf12b..5239562 100644 --- a/tests/ServerMonitorManager.Control.Tests/ControlStoreTests.cs +++ b/tests/ServerMonitorManager.Control.Tests/ControlStoreTests.cs @@ -54,8 +54,13 @@ public sealed class ControlStoreTests : IAsyncDisposable var first = await store.RecordHeartbeatAsync(heartbeat, 30, cancellationToken); var retry = await store.RecordHeartbeatAsync(heartbeat, 30, cancellationToken); - Assert.Equal(first, retry); - Assert.Equal(1, first.Sequence); + Assert.Equal(first.Response, retry.Response); + Assert.True(first.RequiresReconciliation); + Assert.True(retry.RequiresReconciliation); + Assert.Equal(1, first.Response.Sequence); + await store.CompleteAgentReconciliationAsync("home", cancellationToken); + var completedRetry = await store.RecordHeartbeatAsync(heartbeat, 30, cancellationToken); + Assert.False(completedRetry.RequiresReconciliation); await using (var connection = new SqliteConnection( $"Data Source={Path.Combine(_directory, "control.db")}")) { @@ -70,6 +75,118 @@ public sealed class ControlStoreTests : IAsyncDisposable cancellationToken)); } + [Fact] + public async Task ReconnectReappliesOnlyLatestDisabledPolicy() + { + var cancellationToken = TestContext.Current.CancellationToken; + var store = CreateStore(); + await store.InitializeAsync(cancellationToken); + await EnrollAgentAsync(store, "ai-agent", "AB12", cancellationToken); + await EnrollAgentAsync(store, "home", "CD34", cancellationToken); + var broker = new ControlEventBroker(); + var applier = new CountingPolicyApplier(); + var service = new LinkService(store, applier, broker); + + var first = await service.CreateAsync( + new LinkPolicyCreateRequest( + "ai-agent", "home", "tcp", 22, 60, "first", Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + await service.DisableAsync( + first.Id, + new LinkPolicyDisableRequest(Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + _ = await service.CreateAsync( + new LinkPolicyCreateRequest( + "ai-agent", "home", "tcp", 22, 60, "replacement", Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + + var disconnectsBeforeStaleRequest = applier.DisconnectCalls; + var stale = await service.DisableAsync( + first.Id, + new LinkPolicyDisableRequest(Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + Assert.Equal("Disabled", stale?.DesiredState); + Assert.Equal(disconnectsBeforeStaleRequest, applier.DisconnectCalls); + + var activeResult = await service.ReconcileDisabledLinksForNodeAsync("home", cancellationToken); + Assert.Equal(new LinkReconciliationResult(0, 0), activeResult); + + var latest = (await store.ListEffectiveLinksForNodeAsync("home", cancellationToken)).Single(); + await service.DisableAsync( + latest.Id, + new LinkPolicyDisableRequest(Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + var beforeReconnect = applier.DisconnectCalls; + + var disabledResult = await service.ReconcileDisabledLinksForNodeAsync("home", cancellationToken); + + Assert.Equal(new LinkReconciliationResult(1, 0), disabledResult); + Assert.Equal(beforeReconnect + 1, applier.DisconnectCalls); + var persisted = (await store.ListEffectiveLinksForNodeAsync("home", cancellationToken)).Single(); + Assert.Equal("Disabled", persisted.DesiredState); + Assert.Equal("Disabled", persisted.ActualState); + } + + [Fact] + public async Task FailedReconnectRemainsPendingUntilFirewallSucceeds() + { + var cancellationToken = TestContext.Current.CancellationToken; + var store = CreateStore(); + await store.InitializeAsync(cancellationToken); + await EnrollAgentAsync(store, "ai-agent", "EF56", cancellationToken); + await EnrollAgentAsync(store, "home", "7890", cancellationToken); + var applier = new CountingPolicyApplier(); + var service = new LinkService(store, applier, new ControlEventBroker()); + var active = await service.CreateAsync( + new LinkPolicyCreateRequest( + "ai-agent", "home", "tcp", 22, 60, "test", Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + await service.DisableAsync( + active.Id, + new LinkPolicyDisableRequest(Guid.NewGuid().ToString()), + "windows-pc", + cancellationToken); + + var heartbeat = new AgentHeartbeat( + "home", "test", DateTimeOffset.UtcNow, 0.5, 1, 2, 3, 4, 5, 6, 7, + Guid.NewGuid().ToString()); + var first = await store.RecordHeartbeatAsync(heartbeat, 30, cancellationToken); + Assert.True(first.RequiresReconciliation); + + applier.FailDisconnect = true; + var failed = await service.ReconcileDisabledLinksForNodeAsync("home", cancellationToken); + Assert.Equal(new LinkReconciliationResult(1, 1), failed); + var retry = await store.RecordHeartbeatAsync( + heartbeat with + { + SentAt = DateTimeOffset.UtcNow, + IdempotencyKey = Guid.NewGuid().ToString() + }, + 30, + cancellationToken); + Assert.True(retry.RequiresReconciliation); + + applier.FailDisconnect = false; + var succeeded = await service.ReconcileDisabledLinksForNodeAsync("home", cancellationToken); + Assert.Equal(new LinkReconciliationResult(1, 0), succeeded); + await store.CompleteAgentReconciliationAsync("home", cancellationToken); + var completed = await store.RecordHeartbeatAsync( + heartbeat with + { + SentAt = DateTimeOffset.UtcNow, + IdempotencyKey = Guid.NewGuid().ToString() + }, + 30, + cancellationToken); + Assert.False(completed.RequiresReconciliation); + } + [Fact] public async Task DeviceCertificateReceivesOperatorIdentity() { @@ -320,4 +437,27 @@ public sealed class ControlStoreTests : IAsyncDisposable Assert.Equal("Disconnecting", persisted.ActualState); } } + + private sealed class CountingPolicyApplier : ILinkPolicyApplier + { + public int ConnectCalls { get; private set; } + public int DisconnectCalls { get; private set; } + public bool FailDisconnect { get; set; } + + public Task ApplyConnectAsync(LinkPolicy link, CancellationToken cancellationToken) + { + ConnectCalls++; + return Task.CompletedTask; + } + + public Task ApplyDisconnectAsync(LinkPolicy link, CancellationToken cancellationToken) + { + DisconnectCalls++; + if (FailDisconnect) + { + throw new InvalidOperationException("simulated firewall failure"); + } + return Task.CompletedTask; + } + } }