From 64d7519837f5075fcbc260197f0ce426937f4c24 Mon Sep 17 00:00:00 2001 From: IgorRachkov <89467086+igor04091968@users.noreply.github.com> Date: Tue, 5 May 2026 00:05:37 +0300 Subject: [PATCH] Implement Stage 1 reliability transport for DLP collectors --- windows/dlp-endpoint-signals-collector.ps1 | 88 ++++++++++++++++++-- windows/file-operations-collector.ps1 | 97 ++++++++++++++++++++-- 2 files changed, 175 insertions(+), 10 deletions(-) diff --git a/windows/dlp-endpoint-signals-collector.ps1 b/windows/dlp-endpoint-signals-collector.ps1 index 1052c02..cf1f260 100644 --- a/windows/dlp-endpoint-signals-collector.ps1 +++ b/windows/dlp-endpoint-signals-collector.ps1 @@ -1,6 +1,6 @@ [CmdletBinding()] param( - [string]$ConfigPath = 'C:\ProgramData\ActivityWatch\deployment-config.json', + [string]$ConfigPath = 'C:\ProgramData\AWatch-rus\deployment-config.json', [string]$ServerHost, [int]$ServerPort, [ValidateSet('http', 'https')] @@ -33,6 +33,66 @@ function Write-EndpointLog { } } + +function Get-NewEventId { return ([guid]::NewGuid().ToString()) } + +function Initialize-TransportQueue { + param([Parameter(Mandatory = $true)][string]$QueuePath) + $script:QueuePath = $QueuePath + try { + $dir = Split-Path -Path $QueuePath -Parent + if ($dir -and -not (Test-Path -LiteralPath $dir)) { New-Item -ItemType Directory -Path $dir -Force | Out-Null } + if (-not (Test-Path -LiteralPath $QueuePath)) { New-Item -ItemType File -Path $QueuePath -Force | Out-Null } + } + catch { + Write-EndpointLog ("Queue init error: {0}" -f $_.Exception.Message) + } +} + +function Add-TransportQueueRecord { + param([string]$Uri,[string]$Json) + if (-not $script:QueuePath) { return } + $record = @{ id = (Get-NewEventId); createdAt = (Get-Date).ToUniversalTime().ToString('o'); uri = $Uri; payload = $Json } + Add-Content -LiteralPath $script:QueuePath -Value ($record | ConvertTo-Json -Compress) + $script:TransportStats.eventsEnqueued++ +} + +function Get-QueueDepth { + if (-not $script:QueuePath -or -not (Test-Path -LiteralPath $script:QueuePath)) { return 0 } + return @((Get-Content -LiteralPath $script:QueuePath)).Count +} + +function Send-WithQueue { + param([string]$Uri,[string]$Json) + Add-TransportQueueRecord -Uri $Uri -Json $Json + Try-FlushTransportQueue -MaxItems 20 +} + +function Try-FlushTransportQueue { + param([int]$MaxItems = 20) + if (-not $script:QueuePath -or -not (Test-Path -LiteralPath $script:QueuePath)) { return } + $lines = @(Get-Content -LiteralPath $script:QueuePath) + if ($lines.Count -eq 0) { return } + $remaining = New-Object System.Collections.Generic.List[string] + $sent = 0 + foreach ($line in $lines) { + if ($sent -ge $MaxItems) { $remaining.Add($line); continue } + try { $rec = $line | ConvertFrom-Json } catch { $remaining.Add($line); continue } + if (Invoke-AwJsonPost -Uri ([string]$rec.uri) -Json ([string]$rec.payload)) { + $script:TransportStats.eventsSent++ + $script:TransportStats.lastSendStatus = 'ok' + $sent++ + } + else { + $script:TransportStats.sendFailures++ + $script:TransportStats.lastSendStatus = 'failed' + $remaining.Add($line) + break + } + } + Set-Content -LiteralPath $script:QueuePath -Value $remaining +} + function Invoke-AwJsonPost { param( [Parameter(Mandatory = $true)][string]$Uri, @@ -40,7 +100,14 @@ function Invoke-AwJsonPost { ) $bytes = [Text.Encoding]::UTF8.GetBytes($Json) - Invoke-RestMethod -Method Post -Uri $Uri -ContentType 'application/json; charset=utf-8' -Body $bytes | Out-Null + try { + Invoke-RestMethod -Method Post -Uri $Uri -ContentType 'application/json; charset=utf-8' -Body $bytes | Out-Null + return $true + } + catch { + Write-EndpointLog ("POST Error: {0}" -f $_.Exception.Message) + return $false + } } function Ensure-Bucket { @@ -60,7 +127,7 @@ function Ensure-Bucket { hostname = $script:Hostname } | ConvertTo-Json -Compress - Invoke-AwJsonPost -Uri "$($script:ApiBase)/buckets/$BucketId" -Json $body + Send-WithQueue -Uri "$($script:ApiBase)/buckets/$BucketId" -Json $body $script:KnownBuckets[$BucketId] = $true } @@ -77,6 +144,8 @@ function Send-EndpointSignalHeartbeat { timestamp = (Get-Date).ToUniversalTime().ToString('yyyy-MM-ddTHH:mm:ss.fffZ') duration = 0 data = @{ + eventId = (Get-NewEventId) + eventCreatedAt = (Get-Date).ToUniversalTime().ToString('o') signalType = $SignalType username = $env:USERNAME sessionId = $script:SessionId @@ -85,7 +154,7 @@ function Send-EndpointSignalHeartbeat { } + $Data } | ConvertTo-Json -Depth 6 -Compress - Invoke-AwJsonPost -Uri "$($script:ApiBase)/buckets/$bucketId/heartbeat?pulsetime=$script:PulseSeconds" -Json $payload + Send-WithQueue -Uri "$($script:ApiBase)/buckets/$bucketId/heartbeat?pulsetime=$script:PulseSeconds" -Json $payload } function Send-DlpIncidentHeartbeat { @@ -114,6 +183,8 @@ function Send-DlpIncidentHeartbeat { timestamp = (Get-Date).ToUniversalTime().ToString('yyyy-MM-ddTHH:mm:ss.fffZ') duration = 0 data = @{ + eventId = (Get-NewEventId) + eventCreatedAt = (Get-Date).ToUniversalTime().ToString('o') ruleId = $RuleId action = $Action severity = $Severity @@ -126,7 +197,7 @@ function Send-DlpIncidentHeartbeat { } + $Data + $captureData } | ConvertTo-Json -Depth 7 -Compress - Invoke-AwJsonPost -Uri "$($script:ApiBase)/buckets/$bucketId/heartbeat?pulsetime=$script:PulseSeconds" -Json $payload + Send-WithQueue -Uri "$($script:ApiBase)/buckets/$bucketId/heartbeat?pulsetime=$script:PulseSeconds" -Json $payload } function Get-FileSha256Hex { @@ -760,6 +831,11 @@ $script:LogPath = $resolvedLogPath $script:IncidentArtifactsRoot = $resolvedIncidentArtifactsRoot $script:IncidentScreenshotEnabled = $resolvedIncidentScreenshotEnabled $script:ScreenshotTypesLoaded = $false +$script:TransportStats = @{ eventsEnqueued = 0; eventsSent = 0; sendFailures = 0; lastSendStatus = 'init' } +$script:QueuePath = $null + +$queueFile = Join-Path $resolvedLogsRoot ("endpoint-queue-{0}.jsonl" -f $env:USERNAME) +Initialize-TransportQueue -QueuePath $queueFile Load-DlpPolicy -Path $resolvedPolicyPath Write-EndpointLog ("endpoint collector started against {0}" -f $script:ApiBase) @@ -906,5 +982,7 @@ while ($true) { Write-EndpointLog ("collector error: {0}" -f $_.Exception.Message) } + Try-FlushTransportQueue -MaxItems 50 + Write-EndpointLog ("transport metrics queueDepth={0} enqueued={1} sent={2} failures={3} lastStatus={4}" -f (Get-QueueDepth), $script:TransportStats.eventsEnqueued, $script:TransportStats.eventsSent, $script:TransportStats.sendFailures, $script:TransportStats.lastSendStatus) Start-Sleep -Seconds $resolvedPollSeconds } diff --git a/windows/file-operations-collector.ps1 b/windows/file-operations-collector.ps1 index 3fb5246..7e0997f 100644 --- a/windows/file-operations-collector.ps1 +++ b/windows/file-operations-collector.ps1 @@ -26,6 +26,8 @@ $script:SessionId = [System.Diagnostics.Process]::GetCurrentProcess().SessionId # Настройка логирования $script:LogPath = $LogPath $script:LocalAgentLogsEnabled = [bool]$LogPath +$script:TransportStats = @{ eventsEnqueued = 0; eventsSent = 0; sendFailures = 0; lastSendStatus = 'init' } +$script:QueuePath = $null function Get-DeploymentConfig { param([string]$Path) @@ -43,19 +45,89 @@ function Write-FileCollectorLog { } catch {} } + +function Get-NewEventId { return ([guid]::NewGuid().ToString()) } + +function Initialize-TransportQueue { + param([Parameter(Mandatory = $true)][string]$QueuePath) + $script:QueuePath = $QueuePath + try { + $dir = Split-Path -Path $QueuePath -Parent + if ($dir -and -not (Test-Path -LiteralPath $dir)) { New-Item -ItemType Directory -Path $dir -Force | Out-Null } + if (-not (Test-Path -LiteralPath $QueuePath)) { New-Item -ItemType File -Path $QueuePath -Force | Out-Null } + } catch { + Write-FileCollectorLog "Queue init error: $($_.Exception.Message)" + } +} + +function Add-TransportQueueRecord { + param([string]$Uri,[string]$Json) + if (-not $script:QueuePath) { return } + $record = @{ id = (Get-NewEventId); createdAt = (Get-Date).ToUniversalTime().ToString('o'); uri = $Uri; payload = $Json } + Add-Content -LiteralPath $script:QueuePath -Value ($record | ConvertTo-Json -Compress) + $script:TransportStats.eventsEnqueued++ +} + +function Get-QueueDepth { + if (-not $script:QueuePath -or -not (Test-Path -LiteralPath $script:QueuePath)) { return 0 } + return @((Get-Content -LiteralPath $script:QueuePath)).Count +} + +function Try-FlushTransportQueue { + param([int]$MaxItems = 20) + if (-not $script:QueuePath -or -not (Test-Path -LiteralPath $script:QueuePath)) { return } + $lines = @(Get-Content -LiteralPath $script:QueuePath) + if ($lines.Count -eq 0) { return } + $remaining = New-Object System.Collections.Generic.List[string] + $sentInRun = 0 + foreach ($line in $lines) { + if ($sentInRun -ge $MaxItems) { $remaining.Add($line); continue } + try { $rec = $line | ConvertFrom-Json } catch { $remaining.Add($line); continue } + if (Invoke-AwJsonPost -Uri ([string]$rec.uri) -Json ([string]$rec.payload)) { + $script:TransportStats.eventsSent++ + $script:TransportStats.lastSendStatus = 'ok' + $sentInRun++ + } else { + $script:TransportStats.sendFailures++ + $script:TransportStats.lastSendStatus = 'failed' + $remaining.Add($line) + break + } + } + if ($sentInRun -lt $lines.Count) { + for ($i=$sentInRun+($lines.Count-$remaining.Count); $i -lt $lines.Count; $i++) { } + } + Set-Content -LiteralPath $script:QueuePath -Value $remaining +} + +function Send-WithQueue { + param([string]$Uri,[string]$Json) + Add-TransportQueueRecord -Uri $Uri -Json $Json + Try-FlushTransportQueue -MaxItems 10 +} + function Invoke-AwJsonPost { param( [Parameter(Mandatory = $true)][string]$Uri, [Parameter(Mandatory = $true)][string]$Json ) + $httpClient = New-Object System.Net.Http.HttpClient try { - $httpClient = New-Object System.Net.Http.HttpClient $content = New-Object System.Net.Http.StringContent($Json, [System.Text.Encoding]::UTF8, "application/json") $response = $httpClient.PostAsync($Uri, $content).Result - $httpClient.Dispose() + if (-not $response.IsSuccessStatusCode) { + $statusCode = [int]$response.StatusCode + $reason = [string]$response.ReasonPhrase + $responseBody = $response.Content.ReadAsStringAsync().Result + Write-FileCollectorLog ("POST failed uri={0} status={1} reason={2} body={3}" -f $Uri, $statusCode, $reason, $responseBody) + } } catch { Write-FileCollectorLog "POST Error: $($_.Exception.Message)" + return $false + } finally { + $httpClient.Dispose() } + return $response.IsSuccessStatusCode } function Ensure-Bucket { @@ -70,7 +142,7 @@ function Ensure-Bucket { type = $BucketType hostname = $script:Hostname } | ConvertTo-Json -Compress - Invoke-AwJsonPost -Uri "$($script:ApiBase)/buckets/$BucketId" -Json $body + Send-WithQueue -Uri "$($script:ApiBase)/buckets/$BucketId" -Json $body $script:KnownBuckets[$BucketId] = $true } @@ -86,6 +158,9 @@ function Send-FileOperationEvent { Ensure-Bucket -BucketId $bucketId -ClientName 'aw-file-operations' -BucketType 'aw.file.operation' $data = @{ + eventId = (Get-NewEventId) + eventCreatedAt = (Get-Date).ToUniversalTime().ToString('o') + operation = $Operation path = $FilePath extension = [System.IO.Path]::GetExtension($FilePath) @@ -106,7 +181,7 @@ function Send-FileOperationEvent { data = $data } | ConvertTo-Json -Depth 5 -Compress - Invoke-AwJsonPost -Uri "$($script:ApiBase)/buckets/$bucketId/heartbeat?pulsetime=15" -Json $payload + Send-WithQueue -Uri "$($script:ApiBase)/buckets/$bucketId/heartbeat?pulsetime=15" -Json $payload } $config = Get-DeploymentConfig -Path $ConfigPath @@ -117,6 +192,9 @@ $hostName = if ($ServerHost) { $ServerHost } elseif ($config.server.host) { $con $port = if ($ServerPort) { $ServerPort } elseif ($config.server.port) { $config.server.port } else { 5600 } $script:ApiBase = "{0}://{1}:{2}/api/0" -f $scheme, $hostName, $port +$queueFile = Join-Path ([System.IO.Path]::GetDirectoryName($script:LogPath)) ("file-collector-queue-{0}.jsonl" -f $env:USERNAME) +Initialize-TransportQueue -QueuePath $queueFile + $bucketId = 'aw-file-operations_' + $script:Hostname Ensure-Bucket -BucketId $bucketId -ClientName 'aw-file-operations' -BucketType 'aw.file.operation' @@ -144,6 +222,7 @@ if ($resolvedPaths.Count -eq 0) { Write-FileCollectorLog "Starting watch on paths: $($resolvedPaths -join ', ')" $watchers = @() +$subscriptions = @() foreach ($path in $resolvedPaths) { $watcher = New-Object System.IO.FileSystemWatcher $watcher.Path = $path @@ -162,7 +241,7 @@ foreach ($path in $resolvedPaths) { $onRenamed = Register-ObjectEvent $watcher "Renamed" -Action { Send-FileOperationEvent -Operation 'Renamed' -FilePath $Event.SourceEventArgs.FullPath -OldFilePath $Event.SourceEventArgs.OldFullPath } - + $subscriptions += @($onChanged, $onDeleted, $onRenamed) $watchers += $watcher } @@ -170,11 +249,19 @@ Write-FileCollectorLog "Collector started. Waiting for events..." try { while ($true) { + Try-FlushTransportQueue -MaxItems 50 + Write-FileCollectorLog ("transport metrics queueDepth={0} enqueued={1} sent={2} failures={3} lastStatus={4}" -f (Get-QueueDepth), $script:TransportStats.eventsEnqueued, $script:TransportStats.eventsSent, $script:TransportStats.sendFailures, $script:TransportStats.lastSendStatus) Start-Sleep -Seconds $PollSeconds } } finally { Write-FileCollectorLog "Stopping collector..." + foreach ($sub in @($subscriptions)) { + if ($null -ne $sub) { + try { Unregister-Event -SubscriptionId $sub.Id -ErrorAction SilentlyContinue } catch {} + try { Remove-Job -Id $sub.Id -Force -ErrorAction SilentlyContinue } catch {} + } + } foreach ($w in $watchers) { $w.EnableRaisingEvents = $false $w.Dispose()