From e8a8b2f926b3b21947e6e9c0c78697645ff0378d Mon Sep 17 00:00:00 2001 From: Rogee Date: Tue, 22 Sep 2026 13:02:55 +0800 Subject: [PATCH] Stop data sync after account authorization revoke --- .../validation/raw/collect-live-operations.sh | 21 ++++- .../raw/verify-p4-live-operations.sh | 2 +- .../WxAgent.Core/RemoteControlClient.cs | 24 +++++- node-agent/WxAgent.Core/RemoteDataSync.cs | 11 +++ .../RemoteAgentHostedService.cs | 36 +++++++-- .../WxAgent.Core.Tests/RemoteControlTests.cs | 77 +++++++++++++++++++ 6 files changed, 159 insertions(+), 12 deletions(-) diff --git a/docs/validation/raw/collect-live-operations.sh b/docs/validation/raw/collect-live-operations.sh index 5a7fd9d..6bf44e8 100755 --- a/docs/validation/raw/collect-live-operations.sh +++ b/docs/validation/raw/collect-live-operations.sh @@ -125,6 +125,18 @@ after=$(sync_summary) echo "{\"event\":\"agent-restart\",\"process\":$restart_process,\"before\":$before,\"after\":$after}" run_ps "$stop_ps" >/dev/null 2>&1 +# Create a real, valid local pending batch while the control plane is unavailable. +# The agent must later discard this queued body after authorization is revoked. +stop_server +run_ps "Remove-Item \"$WINDOWS_DIR/data/remote-data-sync-state.json\",\"$WINDOWS_DIR/data/remote-data-queue.json\" -Force -ErrorAction SilentlyContinue" >/dev/null 2>&1 +pending_process=$(start_tray) +pending_queue=$(queue_json) +run_ps "$stop_ps" >/dev/null 2>&1 +start_server +for _ in $(seq 1 20); do + sleep 1 + curl -fsS http://127.0.0.1:8090/healthz >/dev/null 2>&1 && break +done web_token=$(curl -fsS -X POST http://127.0.0.1:8090/v1/auth/login -H 'Content-Type: application/json' -d "{\"username\":\"admin\",\"password\":\"$WEB_PASSWORD\"}" | jq -r .access_token) revoke=$(curl -sS -o "$root/revoke" -w '%{http_code}' -X POST -H "Authorization: Bearer $web_token" "http://127.0.0.1:8090/v1/data/accounts/$ACCOUNT_ID/revoke") conv=$(curl -sS -o "$root/conv" -w '%{http_code}' -H "Authorization: Bearer $web_token" "http://127.0.0.1:8090/v1/data/accounts/$ACCOUNT_ID/conversations") @@ -132,7 +144,14 @@ msgs=$(curl -sS -o "$root/msg" -w '%{http_code}' -H "Authorization: Bearer $web_ sync=$(curl -sS -o "$root/sync" -w '%{http_code}' -H "Authorization: Bearer $web_token" "http://127.0.0.1:8090/v1/data/accounts/$ACCOUNT_ID/sync-status") reject_payload=$(jq -nc --arg account "$ACCOUNT_ID" '{node_id:"local-node",account_id:$account,batch_id:"revoked-batch-harness",source_generation:$account,stream_key:"messages",sequence:5,cursor_start:"4",cursor_end:"5",payload_hash:"revoked-batch-hash",coverage_state:"complete",conversations:[],messages:[{message_id:"revoked-message-harness",chat_id:"filehelper",chat_type:"Private",source_message_id:"revoked-source-harness",direction:"incoming",message_type:"text",text:"fixture-revoked",source_time:"2026-09-22T03:00:00Z",observed_at:"2026-09-22T03:00:00Z",source_version:"harness",payload_hash:"revoked-message-hash"}]}') batch_rejected=$(curl -sS -o "$root/rejected" -w '%{http_code}' -X POST -H "Authorization: Bearer $NODE_TOKEN" -H 'Content-Type: application/json' --data-binary "$reject_payload" http://127.0.0.1:8090/v1/data/batches) -echo "{\"event\":\"authorization-revoke\",\"revoke\":$revoke,\"conversations\":$conv,\"messages\":$msgs,\"sync_status\":$sync,\"batch_rejected\":$batch_rejected}" +# Start the agent only after revocation. Its reconciliation/flush path must drop the +# pre-existing queue and keep it empty instead of collecting or retransmitting. +revoked_process=$(start_tray) +revoked_queue=$(queue_json) +sleep 12 +revoked_queue_after=$(queue_json) +echo "{\"event\":\"authorization-revoke\",\"pending_process\":$pending_process,\"pending_queue\":$pending_queue,\"revoke\":$revoke,\"conversations\":$conv,\"messages\":$msgs,\"sync_status\":$sync,\"batch_rejected\":$batch_rejected,\"revoked_process\":$revoked_process,\"revoked_queue\":$revoked_queue,\"revoked_queue_after\":$revoked_queue_after}" +run_ps "$stop_ps" >/dev/null 2>&1 restore_process=$(start_tray) sleep 5 web_token=$(curl -fsS -X POST http://127.0.0.1:8090/v1/auth/login -H 'Content-Type: application/json' -d "{\"username\":\"admin\",\"password\":\"$WEB_PASSWORD\"}" | jq -r .access_token) diff --git a/docs/validation/raw/verify-p4-live-operations.sh b/docs/validation/raw/verify-p4-live-operations.sh index abfb879..7554c76 100755 --- a/docs/validation/raw/verify-p4-live-operations.sh +++ b/docs/validation/raw/verify-p4-live-operations.sh @@ -17,7 +17,7 @@ jq -s -e ' ($e["offline-queue"].queue.queue_pending > 0 and $e["offline-queue"].queue.queue_bytes > 0 and $e["offline-queue"].process.session == 1) and ($e["replay"].queue.queue_pending == 0 and $e["replay"].sync.state == "complete" and $e["replay"].shard.coverage == "complete" and $e["replay"].shard.integrity == "ok") and ($e["agent-restart"].process.session == 1 and $e["agent-restart"].before.confirmed_sequence == $e["agent-restart"].after.confirmed_sequence and $e["agent-restart"].after.state == "complete") and - ($e["authorization-revoke"].revoke == 200 and $e["authorization-revoke"].conversations == 403 and $e["authorization-revoke"].messages == 403 and $e["authorization-revoke"].sync_status == 200 and $e["authorization-revoke"].batch_rejected == 403) and + ($e["authorization-revoke"].pending_process.session == 1 and $e["authorization-revoke"].pending_queue.queue_pending > 0 and $e["authorization-revoke"].pending_queue.queue_bytes > 0 and $e["authorization-revoke"].revoke == 200 and $e["authorization-revoke"].conversations == 403 and $e["authorization-revoke"].messages == 403 and $e["authorization-revoke"].sync_status == 200 and $e["authorization-revoke"].batch_rejected == 403 and $e["authorization-revoke"].revoked_process.session == 1 and $e["authorization-revoke"].revoked_queue.queue_pending == 0 and $e["authorization-revoke"].revoked_queue_after.queue_pending == 0) and ($e["authorization-restore"].process.session == 1 and $e["authorization-restore"].conversations == 200 and $e["authorization-restore"].messages == 200 and $e["authorization-restore"].queue.queue_pending == 0) and ($e["control-plane-restart"].before.integrity == "ok" and $e["control-plane-restart"].after.integrity == "ok" and $e["control-plane-restart"].before.messages == $e["control-plane-restart"].after.messages and $e["control-plane-restart"].before.batches == $e["control-plane-restart"].after.batches) ' "$file" >/dev/null diff --git a/node-agent/WxAgent.Core/RemoteControlClient.cs b/node-agent/WxAgent.Core/RemoteControlClient.cs index 5ce1c8b..67c6eeb 100644 --- a/node-agent/WxAgent.Core/RemoteControlClient.cs +++ b/node-agent/WxAgent.Core/RemoteControlClient.cs @@ -190,24 +190,42 @@ public sealed class RemoteControlClient : IDisposable cancellationToken, RemoteDataProtocol.MaxBatchBytes).ConfigureAwait(false); } + public Task> FlushDataBatchesAsync( + RemoteDataBatchQueue queue, + ReportingConfig reportingConfig, + CancellationToken cancellationToken = default) => + FlushDataBatchesAsync(queue, reportingConfig, new HashSet(StringComparer.Ordinal), cancellationToken); + public async Task> FlushDataBatchesAsync( RemoteDataBatchQueue queue, ReportingConfig reportingConfig, + ISet blockedAccountIds, CancellationToken cancellationToken = default) { EnsureAuthenticated(); var confirmed = new List(); foreach (var batch in queue.Pending()) { + if (blockedAccountIds.Contains(batch.AccountId)) + continue; var filtered = RemoteDataBatchAuthorization.Filter(reportingConfig, batch, out _); if (filtered is null) { queue.Drop(batch.BatchId); continue; } - var acknowledgement = await SubmitDataBatchAsync(reportingConfig, filtered, cancellationToken).ConfigureAwait(false); - if (acknowledgement.Accepted && queue.MarkConfirmed(acknowledgement)) - confirmed.Add(filtered); + try + { + var acknowledgement = await SubmitDataBatchAsync(reportingConfig, filtered, cancellationToken).ConfigureAwait(false); + if (acknowledgement.Accepted && queue.MarkConfirmed(acknowledgement)) + confirmed.Add(filtered); + } + catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized") + { + // Authorization revocation is terminal for locally buffered content: do not retain or retry it. + queue.DropAllForAccount(batch.AccountId); + blockedAccountIds.Add(batch.AccountId); + } } return confirmed; } diff --git a/node-agent/WxAgent.Core/RemoteDataSync.cs b/node-agent/WxAgent.Core/RemoteDataSync.cs index 0fb8232..c17d306 100644 --- a/node-agent/WxAgent.Core/RemoteDataSync.cs +++ b/node-agent/WxAgent.Core/RemoteDataSync.cs @@ -362,6 +362,17 @@ public sealed class RemoteDataBatchQueue } } + public int DropAllForAccount(string accountId) + { + lock (gate) + { + var removed = state.Items.RemoveAll(item => string.Equals(item.AccountId, accountId, StringComparison.Ordinal)); + if (removed > 0) + SaveLocked(); + return removed; + } + } + public bool MarkConfirmed(RemoteDataBatchAck acknowledgement) { lock (gate) diff --git a/node-agent/WxAgent.Service/RemoteAgentHostedService.cs b/node-agent/WxAgent.Service/RemoteAgentHostedService.cs index 1af1e35..48a6fee 100644 --- a/node-agent/WxAgent.Service/RemoteAgentHostedService.cs +++ b/node-agent/WxAgent.Service/RemoteAgentHostedService.cs @@ -48,6 +48,7 @@ public sealed class RemoteAgentHostedService( var syncState = options.EnableDataSync ? new RemoteDataSyncStateStore(Path.Combine(options.DataDirectory, "remote-data-sync-state.json")) : null; + var blockedDataSyncAccounts = new HashSet(StringComparer.Ordinal); var lastDataSyncAt = DateTimeOffset.MinValue; string? registeredActiveAccountId = null; bool? registeredActiveAccountVerified = null; @@ -78,6 +79,7 @@ public sealed class RemoteAgentHostedService( registeredActiveAccountId = snapshot.ActiveAccountId; registeredActiveAccountVerified = snapshot.ActiveAccountVerified; registeredReportingConfigVersion = reporting.ConfigVersion; + blockedDataSyncAccounts.Clear(); retry = RetryDelay; } @@ -87,20 +89,24 @@ public sealed class RemoteAgentHostedService( if (dataQueue is not null && syncState is not null && dataCollector is not null && DateTimeOffset.UtcNow - lastDataSyncAt >= TimeSpan.FromSeconds(options.DataSyncIntervalSeconds)) { - await ReconcileDataStatusAsync(client, reporting, dataQueue, syncState, stoppingToken); + await ReconcileDataStatusAsync(client, reporting, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken); + var blockedBeforeFlush = blockedDataSyncAccounts.ToHashSet(StringComparer.Ordinal); try { - var confirmedBatches = await client.FlushDataBatchesAsync(dataQueue, reporting, stoppingToken); + var confirmedBatches = await client.FlushDataBatchesAsync(dataQueue, reporting, blockedDataSyncAccounts, stoppingToken); foreach (var confirmedBatch in confirmedBatches) syncState.MarkConfirmed(confirmedBatch); + foreach (var accountId in blockedDataSyncAccounts.Except(blockedBeforeFlush, StringComparer.Ordinal)) + logger.LogWarning("Data sync authorization was revoked; accountId={AccountId}; pending content was discarded and collection is paused.", accountId); } catch (Exception exception) when (exception is HttpRequestException or RemoteClientException) { logger.LogWarning("Data sync delivery deferred; type={ExceptionType}.", exception.GetType().Name); } // Collection must continue while the platform is unreachable; the durable queue is - // the offline buffer and will be flushed on the next successful connection. - await CollectDataBatchesAsync(remote, reporting, dataCollector, dataQueue, syncState, stoppingToken); + // the offline buffer and will be flushed on the next successful connection. Revoked + // accounts are the exception: their pending content is discarded and collection stops. + await CollectDataBatchesAsync(remote, reporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken); lastDataSyncAt = DateTimeOffset.UtcNow; } var pollAccountIds = reporting.Accounts @@ -133,14 +139,14 @@ public sealed class RemoteAgentHostedService( { logger.LogWarning("Remote control-plane request failed; code={Code}; status={StatusCode}; correlationId={CorrelationId}.", exception.Code, exception.StatusCode, exception.CorrelationId); - await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, stoppingToken); + await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken); await DelayAsync(retry, stoppingToken); retry = TimeSpan.FromSeconds(Math.Min(retry.TotalSeconds * 2, 30)); } catch (Exception exception) { logger.LogWarning("Remote agent cycle failed; type={ExceptionType}.", exception.GetType().Name); - await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, stoppingToken); + await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken); await DelayAsync(retry, stoppingToken); retry = TimeSpan.FromSeconds(Math.Min(retry.TotalSeconds * 2, 30)); } @@ -153,11 +159,14 @@ public sealed class RemoteAgentHostedService( IRemoteDataCollector collector, RemoteDataBatchQueue queue, RemoteDataSyncStateStore state, + ISet blockedAccountIds, CancellationToken cancellationToken) { foreach (var account in reporting.Accounts.Where(item => item.Enabled)) { cancellationToken.ThrowIfCancellationRequested(); + if (blockedAccountIds.Contains(account.AccountId)) + continue; var scopes = AuthorizedChatScopes(reporting, account.AccountId); if (scopes.Count == 0 || queue.Pending().Any(batch => string.Equals(batch.AccountId, account.AccountId, StringComparison.Ordinal) && string.Equals(batch.StreamKey, "messages", StringComparison.Ordinal))) @@ -191,13 +200,14 @@ public sealed class RemoteAgentHostedService( IRemoteDataCollector? collector, RemoteDataBatchQueue? queue, RemoteDataSyncStateStore? state, + ISet blockedAccountIds, CancellationToken cancellationToken) { if (reporting is null || collector is null || queue is null || state is null) return; try { - await CollectDataBatchesAsync(remote, reporting, collector, queue, state, cancellationToken).ConfigureAwait(false); + await CollectDataBatchesAsync(remote, reporting, collector, queue, state, blockedAccountIds, cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { @@ -214,6 +224,7 @@ public sealed class RemoteAgentHostedService( ReportingConfig reporting, RemoteDataBatchQueue queue, RemoteDataSyncStateStore state, + ISet blockedAccountIds, CancellationToken cancellationToken) { foreach (var account in reporting.Accounts.Where(item => item.Enabled)) @@ -229,12 +240,23 @@ public sealed class RemoteAgentHostedService( // Older control planes do not expose the reconciliation endpoint; retain the legacy ACK path. continue; } + catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized") + { + var droppedOnRevoke = queue.DropAllForAccount(account.AccountId); + blockedAccountIds.Add(account.AccountId); + logger.LogWarning( + "Data sync authorization is revoked; accountId={AccountId}; pending content was discarded and collection is paused; droppedPendingBatches={DroppedPendingBatches}.", + account.AccountId, droppedOnRevoke); + continue; + } catch (HttpRequestException) { // A disconnected platform must not prevent local DB collection. return; } + if (blockedAccountIds.Remove(account.AccountId)) + logger.LogInformation("Data sync authorization was restored; accountId={AccountId}; collection is resumed.", account.AccountId); if (!state.Reconcile(remote, account.AccountId, "messages")) continue; var dropped = queue.DropForAccount(account.AccountId, "messages"); diff --git a/tests/node-agent/WxAgent.Core.Tests/RemoteControlTests.cs b/tests/node-agent/WxAgent.Core.Tests/RemoteControlTests.cs index b31ab70..7e2f070 100644 --- a/tests/node-agent/WxAgent.Core.Tests/RemoteControlTests.cs +++ b/tests/node-agent/WxAgent.Core.Tests/RemoteControlTests.cs @@ -335,6 +335,51 @@ public sealed class RemoteControlClientTests Assert.All(handler.Requests, request => Assert.Equal("Bearer node-token", request.Authorization)); } + [Fact] + public async Task FlushDropsAllQueuedBatchesAndBlocksAccountAfterRevocation() + { + var directory = Directory.CreateTempSubdirectory("wxagent-revoke-"); + try + { + var config = new ReportingConfig + { + Enabled = true, + ConfigVersion = 1, + Accounts = [new AccountReportingConfig + { + AccountId = "account-a", Enabled = true, + AllowedChats = [new AllowedChat { Type = ReportingChatType.Private, ChatId = "chat-a", Enabled = true, IdentityVerified = true }] + }] + }; + var queue = new RemoteDataBatchQueue(Path.Combine(directory.FullName, "queue.json")); + Assert.True(queue.Enqueue(config, SyncBatch("batch-1")).Accepted); + Assert.True(queue.Enqueue(config, SyncBatch("batch-2")).Accepted); + + var handler = new RevokedBatchHandler(); + using var http = new HttpClient(handler); + using var client = new RemoteControlClient(new RemoteAgentOptions + { + AuthAddress = "http://127.0.0.1:8090", + Token = "node-token", + NodeId = "node-1" + }, http); + await client.RegisterAsync(new RemoteNodeRegistration("node-1", "test", RemoteProtocol.Version, ["db-messages"], 1, [])); + + var blockedAccounts = new HashSet(StringComparer.Ordinal); + var confirmed = await client.FlushDataBatchesAsync(queue, config, blockedAccounts); + + Assert.Empty(confirmed); + Assert.Contains("account-a", blockedAccounts); + Assert.Empty(queue.Pending()); + Assert.Empty(new RemoteDataBatchQueue(Path.Combine(directory.FullName, "queue.json")).Pending()); + Assert.Equal(1, handler.BatchRequests); + } + finally + { + directory.Delete(true); + } + } + [Fact] public async Task MissingRemoteCredentialsAreRejectedWithoutNetworkAccess() { @@ -345,6 +390,38 @@ public sealed class RemoteControlClientTests Assert.Empty(handler.Requests); } + private static RemoteSyncBatch SyncBatch(string batchId) => new( + "node-1", "account-a", batchId, "generation-a", "messages", 1, "{}", "{\"chat-a\\u001fmessage/a.db\":1}", + "hash-" + batchId, "complete", [], + [new RemoteSyncMessage(batchId, "chat-a", ReportingChatType.Private, batchId, "incoming", "text", "queued secret", + DateTimeOffset.UtcNow, DateTimeOffset.UtcNow, "wx-1", "payload-" + batchId)]); + + private sealed class RevokedBatchHandler : HttpMessageHandler + { + public int BatchRequests { get; private set; } + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + if (request.RequestUri!.AbsolutePath == "/v1/data/batches") + { + BatchRequests++; + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.Forbidden) + { + Content = new StringContent("{\"error\":{\"code\":\"AccountNotAuthorized\",\"message\":\"revoked\"}}", Encoding.UTF8, "application/json") + }); + } + if (request.RequestUri.AbsolutePath == "/v1/nodes/register") + { + var body = JsonSerializer.Serialize(new RemoteNodeRegistrationResponse("node-1", RemoteNodeStatus.Online, true, "register"), RemoteJson.Options); + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent(body, Encoding.UTF8, "application/json") + }); + } + throw new InvalidOperationException("Unexpected request " + request.RequestUri.AbsolutePath); + } + } + private sealed class RecordingHandler : HttpMessageHandler { public List Requests { get; } = [];