Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 lussierj
|
8b0d32cd6b | ||
|
lussierj
|
e898f283ab |
No files matched your search
@@ -2,7 +2,10 @@
|
||||
[Parameter(Mandatory=$true)]
|
||||
[string]$IpAddress,
|
||||
|
||||
[string]$MinerId = ""
|
||||
[string]$MinerId = "",
|
||||
[string]$DiscoveryId = "",
|
||||
[string]$ObservationVersion = "",
|
||||
[string]$CommandId = ""
|
||||
)
|
||||
|
||||
$ConfigPath = "C:\MinerOpsAgent\agent-config.json"
|
||||
@@ -23,6 +26,24 @@ function Invoke-MinerCommandRaw {
|
||||
[int]$TimeoutMs = 2500
|
||||
)
|
||||
|
||||
# This transport accepts only the exact queries used by this enrichment script.
|
||||
# Validate before creating a socket; callers cannot reinterpret a write as a read.
|
||||
$readCommands = @("summary", "devs", "edevs", "stats", "pools", "devdetails",
|
||||
"version", "get_version", "get_miner_info")
|
||||
# PowerShell containment operators use culture comparisons, even with -c.
|
||||
# Match ordinally and serialize the canonical literal; never trim/normalize.
|
||||
$approvedCommand = $null
|
||||
foreach ($candidate in $readCommands) {
|
||||
if ([string]::Equals($candidate, $Command, [StringComparison]::Ordinal)) {
|
||||
$approvedCommand = $candidate
|
||||
break
|
||||
}
|
||||
}
|
||||
if ($Port -ne 4028 -or $null -eq $approvedCommand) {
|
||||
throw "unsupported_diagnostic_operation_stage_a"
|
||||
}
|
||||
$requestJson = @{ cmd = $approvedCommand } | ConvertTo-Json -Compress
|
||||
|
||||
$client = $null
|
||||
|
||||
try {
|
||||
@@ -38,7 +59,7 @@ function Invoke-MinerCommandRaw {
|
||||
$client.SendTimeout = $TimeoutMs
|
||||
|
||||
$stream = $client.GetStream()
|
||||
$payload = [System.Text.Encoding]::ASCII.GetBytes("{`"cmd`":`"$Command`"}`n")
|
||||
$payload = [System.Text.Encoding]::ASCII.GetBytes($requestJson + "`n")
|
||||
$stream.Write($payload, 0, $payload.Length)
|
||||
|
||||
try {
|
||||
@@ -146,6 +167,14 @@ $hubUrl = ([string]$config.hub_url).TrimEnd("/")
|
||||
$secret = [string]$config.agent_secret
|
||||
$agentId = [string]$config.agent_id
|
||||
|
||||
# Revalidate the claimed command against current Hub authority before miner sockets.
|
||||
if (!$DiscoveryId -or !$ObservationVersion -or !$CommandId) { throw "discovery_observation_required" }
|
||||
$target = Invoke-RestMethod -Method Post -Uri "$hubUrl/api/agent/discovery-target" `
|
||||
-Headers @{ "X-MinerOps-Agent-ID" = $agentId; "X-MinerOps-Agent-Secret" = $secret } `
|
||||
-ContentType "application/json" -Body (@{ discovery_id=$DiscoveryId; version=$ObservationVersion; command_id=$CommandId } | ConvertTo-Json)
|
||||
$IpAddress = [string]$target.ip_address
|
||||
if (!$target.ok -or !$IpAddress) { throw "discovery_target_denied" }
|
||||
|
||||
$commands = @(
|
||||
"summary",
|
||||
"devs",
|
||||
@@ -201,6 +230,9 @@ if (!$model) {
|
||||
$payload = [ordered]@{
|
||||
ip_address = $IpAddress
|
||||
miner_id = $MinerId
|
||||
discovery_id = $DiscoveryId
|
||||
version = $ObservationVersion
|
||||
command_id = $CommandId
|
||||
mac_address = $mac
|
||||
hostname = $hostname
|
||||
miner_sn = $minerSn
|
||||
|
||||
@@ -58,6 +58,21 @@ function Invoke-AgentApi {
|
||||
"X-MinerOps-Agent-ID" = [string]$config.agent_id
|
||||
}
|
||||
|
||||
$batchField = $null
|
||||
if ($Path -eq "/api/agent/miner-snapshots") { $batchField = "miners" }
|
||||
elseif ($Path -eq "/api/agent/discoveries") { $batchField = "discoveries" }
|
||||
if ($batchField -and $Body -ne $null -and @($Body[$batchField]).Count -gt 128) {
|
||||
$items = @($Body[$batchField])
|
||||
$results = @()
|
||||
for ($offset = 0; $offset -lt $items.Count; $offset += 128) {
|
||||
$batch = @{}
|
||||
foreach ($key in $Body.Keys) { $batch[$key] = $Body[$key] }
|
||||
$last = [Math]::Min($offset + 127, $items.Count - 1)
|
||||
$batch[$batchField] = @($items[$offset..$last])
|
||||
$results += Invoke-AgentApi -Method $Method -Path $Path -Body $batch
|
||||
}
|
||||
return @{ ok = $true; batches = $results; processed = $items.Count }
|
||||
}
|
||||
if ($Body -ne $null) {
|
||||
$json = $Body | ConvertTo-Json -Depth 100 -Compress
|
||||
return Invoke-RestMethod -Method $Method -Uri $uri -Headers $headers -ContentType "application/json" -Body $json -TimeoutSec 180
|
||||
@@ -252,6 +267,7 @@ function Poll-KnownMiners {
|
||||
|
||||
$result = [ordered]@{
|
||||
miner_id = [string]$Miner.miner_id
|
||||
miner_version = [string]$Miner.miner_version
|
||||
ip_address = [string]$Miner.ip_address
|
||||
mac_address = [string]$Miner.mac_address
|
||||
room_name = [string]$Miner.room_name
|
||||
@@ -592,7 +608,7 @@ public static class MinerOpsFastScanner {
|
||||
}
|
||||
|
||||
function Discover-MinersInRanges {
|
||||
$config = Read-AgentConfig
|
||||
$config = Invoke-AgentApi -Method "GET" -Path "/api/agent/config"
|
||||
$ranges = @($config.ip_ranges | Where-Object { $_.enabled -eq $true })
|
||||
|
||||
Ensure-FastScannerLoaded
|
||||
@@ -743,61 +759,7 @@ function Invoke-MinerPlainCommand {
|
||||
|
||||
[int]$TimeoutMs = 2500
|
||||
)
|
||||
|
||||
$client = $null
|
||||
|
||||
try {
|
||||
$client = New-Object System.Net.Sockets.TcpClient
|
||||
$async = $client.BeginConnect($IpAddress, $Port, $null, $null)
|
||||
|
||||
if (-not $async.AsyncWaitHandle.WaitOne($TimeoutMs)) {
|
||||
throw "Connect timeout to $IpAddress`:$Port"
|
||||
}
|
||||
|
||||
$client.EndConnect($async)
|
||||
$client.ReceiveTimeout = $TimeoutMs
|
||||
$client.SendTimeout = $TimeoutMs
|
||||
|
||||
$stream = $client.GetStream()
|
||||
$payload = [System.Text.Encoding]::ASCII.GetBytes("{`"cmd`":`"$Command`"}`n")
|
||||
$stream.Write($payload, 0, $payload.Length)
|
||||
|
||||
try {
|
||||
$client.Client.Shutdown([System.Net.Sockets.SocketShutdown]::Send)
|
||||
} catch {
|
||||
}
|
||||
|
||||
$buffer = New-Object byte[] 65536
|
||||
$chunks = New-Object System.Collections.ArrayList
|
||||
$started = Get-Date
|
||||
|
||||
while (((Get-Date) - $started).TotalMilliseconds -lt $TimeoutMs) {
|
||||
if (-not $stream.DataAvailable) {
|
||||
Start-Sleep -Milliseconds 20
|
||||
continue
|
||||
}
|
||||
|
||||
$read = $stream.Read($buffer, 0, $buffer.Length)
|
||||
|
||||
if ($read -le 0) {
|
||||
break
|
||||
}
|
||||
|
||||
for ($i = 0; $i -lt $read; $i++) {
|
||||
[void]$chunks.Add([byte]$buffer[$i])
|
||||
}
|
||||
|
||||
if ($chunks.Count -gt 1048576) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return [System.Text.Encoding]::UTF8.GetString([byte[]]$chunks.ToArray([byte])).Trim()
|
||||
} finally {
|
||||
if ($client -ne $null) {
|
||||
$client.Close()
|
||||
}
|
||||
}
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
function Get-MinerOpsApi3Credential {
|
||||
@@ -894,61 +856,7 @@ function Invoke-MinerOpsApi3Payload {
|
||||
|
||||
[int]$TimeoutMs = 10000
|
||||
)
|
||||
|
||||
$client = $null
|
||||
$started = Get-Date
|
||||
|
||||
try {
|
||||
$client = New-Object System.Net.Sockets.TcpClient
|
||||
|
||||
$async = $client.BeginConnect($IpAddress, $Port, $null, $null)
|
||||
|
||||
if (-not $async.AsyncWaitHandle.WaitOne($TimeoutMs)) {
|
||||
throw "API3 connect timeout to $IpAddress`:$Port"
|
||||
}
|
||||
|
||||
$client.EndConnect($async)
|
||||
$client.ReceiveTimeout = $TimeoutMs
|
||||
$client.SendTimeout = $TimeoutMs
|
||||
|
||||
$json = $Payload | ConvertTo-Json -Depth 20 -Compress
|
||||
$requestBytes = [System.Text.Encoding]::UTF8.GetBytes($json)
|
||||
$lengthBytes = [System.BitConverter]::GetBytes([UInt32]$requestBytes.Length)
|
||||
|
||||
$stream = $client.GetStream()
|
||||
$stream.Write($lengthBytes, 0, $lengthBytes.Length)
|
||||
$stream.Write($requestBytes, 0, $requestBytes.Length)
|
||||
|
||||
$responseLengthBytes = Read-MinerOpsExactBytes -Stream $stream -Length 4 -TimeoutMs $TimeoutMs
|
||||
$responseLength = [System.BitConverter]::ToUInt32($responseLengthBytes, 0)
|
||||
|
||||
if ($responseLength -le 0 -or $responseLength -gt 1048576) {
|
||||
throw "API3 response length looked invalid: $responseLength"
|
||||
}
|
||||
|
||||
$responseBytes = Read-MinerOpsExactBytes -Stream $stream -Length ([int]$responseLength) -TimeoutMs $TimeoutMs
|
||||
$responseText = [System.Text.Encoding]::UTF8.GetString($responseBytes)
|
||||
$responseJson = $responseText | ConvertFrom-Json
|
||||
|
||||
return @{
|
||||
sent = $true
|
||||
protocol = "whatsminer_api3"
|
||||
target_ip = $IpAddress
|
||||
target_port = $Port
|
||||
duration_seconds = [math]::Round(((Get-Date) - $started).TotalSeconds, 3)
|
||||
request = @{
|
||||
cmd = $Payload.cmd
|
||||
param = $Payload.param
|
||||
account = $Payload.account
|
||||
has_token = [bool]$Payload.token
|
||||
}
|
||||
response = $responseJson
|
||||
}
|
||||
} finally {
|
||||
if ($client -ne $null) {
|
||||
$client.Close()
|
||||
}
|
||||
}
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
function Get-MinerOpsApi3Salt {
|
||||
@@ -989,43 +897,7 @@ function Invoke-MinerOpsApi3Service {
|
||||
[ValidateSet("start", "stop", "restart")]
|
||||
[string]$ServiceAction
|
||||
)
|
||||
|
||||
$account = Get-MinerOpsApi3Credential -Name "api3_account" -DefaultValue ""
|
||||
$password = Get-MinerOpsApi3Credential -Name "api3_password" -DefaultValue ""
|
||||
$portRaw = Get-MinerOpsApi3Credential -Name "api3_port" -DefaultValue "4433"
|
||||
|
||||
try {
|
||||
$port = [int]$portRaw
|
||||
} catch {
|
||||
$port = 4433
|
||||
}
|
||||
|
||||
$command = "set.miner.service"
|
||||
$saltInfo = Get-MinerOpsApi3Salt -IpAddress $IpAddress -Port $port
|
||||
$salt = [string]$saltInfo.salt
|
||||
$timestamp = [int64][DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
|
||||
$token = Get-MinerOpsApi3Token -Command $command -Password $password -Salt $salt -Timestamp $timestamp
|
||||
|
||||
$payload = @{
|
||||
cmd = $command
|
||||
param = $ServiceAction
|
||||
ts = $timestamp
|
||||
token = $token
|
||||
account = $account
|
||||
}
|
||||
|
||||
$result = Invoke-MinerOpsApi3Payload -IpAddress $IpAddress -Port $port -Payload $payload
|
||||
|
||||
$result["salt_response"] = $saltInfo.salt_response
|
||||
$result["api3_command"] = $command
|
||||
$result["api3_param"] = $ServiceAction
|
||||
$result["api3_account"] = $account
|
||||
|
||||
if ([int]$result.response.code -ne 0) {
|
||||
throw "API3 $command $ServiceAction failed: $($result.response | ConvertTo-Json -Compress)"
|
||||
}
|
||||
|
||||
return $result
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
|
||||
@@ -1054,45 +926,7 @@ function Invoke-MinerOpsApi3AuthenticatedCommand {
|
||||
|
||||
[object]$Param = $null
|
||||
)
|
||||
|
||||
$account = Get-MinerOpsApi3Credential -Name "api3_account" -DefaultValue ""
|
||||
$password = Get-MinerOpsApi3Credential -Name "api3_password" -DefaultValue ""
|
||||
$portRaw = Get-MinerOpsApi3Credential -Name "api3_port" -DefaultValue "4433"
|
||||
|
||||
try {
|
||||
$port = [int]$portRaw
|
||||
} catch {
|
||||
$port = 4433
|
||||
}
|
||||
|
||||
$saltInfo = Get-MinerOpsApi3Salt -IpAddress $IpAddress -Port $port
|
||||
$salt = [string]$saltInfo.salt
|
||||
$timestamp = [int64][DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
|
||||
$token = Get-MinerOpsApi3Token -Command $Command -Password $password -Salt $salt -Timestamp $timestamp
|
||||
|
||||
$payload = @{
|
||||
cmd = $Command
|
||||
ts = $timestamp
|
||||
token = $token
|
||||
account = $account
|
||||
}
|
||||
|
||||
if ($null -ne $Param) {
|
||||
$payload["param"] = $Param
|
||||
}
|
||||
|
||||
$result = Invoke-MinerOpsApi3Payload -IpAddress $IpAddress -Port $port -Payload $payload
|
||||
|
||||
$result["salt_response"] = $saltInfo.salt_response
|
||||
$result["api3_command"] = $Command
|
||||
$result["api3_param"] = $Param
|
||||
$result["api3_account"] = $account
|
||||
|
||||
if ([int]$result.response.code -ne 0) {
|
||||
throw "API3 $Command failed: $($result.response | ConvertTo-Json -Compress)"
|
||||
}
|
||||
|
||||
return $result
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
function Get-MinerOpsPayloadValue {
|
||||
@@ -1129,113 +963,14 @@ function Invoke-MinerOpsLegacy4028Command {
|
||||
[int] $Port = 4028,
|
||||
[int] $TimeoutMs = 7000
|
||||
)
|
||||
|
||||
$normalizedIp = Normalize-MinerOpsIpAddress -IpAddress $IpAddress
|
||||
|
||||
$client = New-Object System.Net.Sockets.TcpClient
|
||||
$connectResult = $client.BeginConnect($normalizedIp, $Port, $null, $null)
|
||||
|
||||
if (-not $connectResult.AsyncWaitHandle.WaitOne($TimeoutMs, $false)) {
|
||||
try { $client.Close() } catch {}
|
||||
throw "Legacy 4028 connection timed out to $normalizedIp`:$Port"
|
||||
}
|
||||
|
||||
$client.EndConnect($connectResult)
|
||||
|
||||
try {
|
||||
$stream = $client.GetStream()
|
||||
$stream.ReadTimeout = $TimeoutMs
|
||||
$stream.WriteTimeout = $TimeoutMs
|
||||
|
||||
$bytes = [System.Text.Encoding]::ASCII.GetBytes($CommandJson)
|
||||
$stream.Write($bytes, 0, $bytes.Length)
|
||||
$stream.Flush()
|
||||
|
||||
Start-Sleep -Milliseconds 500
|
||||
|
||||
$buffer = New-Object byte[] 65535
|
||||
$builder = New-Object System.Text.StringBuilder
|
||||
|
||||
do {
|
||||
$read = $stream.Read($buffer, 0, $buffer.Length)
|
||||
|
||||
if ($read -gt 0) {
|
||||
[void]$builder.Append([System.Text.Encoding]::ASCII.GetString($buffer, 0, $read))
|
||||
}
|
||||
} while ($stream.DataAvailable)
|
||||
|
||||
$rawResponse = $builder.ToString().Trim([char]0).Trim()
|
||||
|
||||
$parsedResponse = $null
|
||||
|
||||
try {
|
||||
if ($rawResponse) {
|
||||
$parsedResponse = $rawResponse | ConvertFrom-Json
|
||||
}
|
||||
} catch {
|
||||
$parsedResponse = $null
|
||||
}
|
||||
|
||||
return @{
|
||||
success = $true
|
||||
control_method = "legacy_4028"
|
||||
port = $Port
|
||||
request_json = $CommandJson
|
||||
raw_response = $rawResponse
|
||||
parsed_response = $parsedResponse
|
||||
}
|
||||
}
|
||||
finally {
|
||||
try { $stream.Close() } catch {}
|
||||
try { $client.Close() } catch {}
|
||||
}
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
function Invoke-MinerOpsM30Legacy4028Test {
|
||||
param(
|
||||
[Parameter(Mandatory=$true)] $Payload
|
||||
)
|
||||
|
||||
if ($Payload -is [string]) {
|
||||
$Payload = $Payload | ConvertFrom-Json
|
||||
}
|
||||
|
||||
$ipAddress = Get-MinerOpsPayloadValue -Payload $Payload -Name "ip_address"
|
||||
|
||||
if (-not $ipAddress) {
|
||||
$ipAddress = Get-MinerOpsPayloadValue -Payload $Payload -Name "current_ip"
|
||||
}
|
||||
|
||||
if (-not $ipAddress) {
|
||||
throw "Legacy 4028 test failed: no IP address in payload."
|
||||
}
|
||||
|
||||
$legacyJson = Get-MinerOpsPayloadValue -Payload $Payload -Name "legacy_4028_json"
|
||||
$legacyCommand = Get-MinerOpsPayloadValue -Payload $Payload -Name "legacy_4028_command"
|
||||
|
||||
if (-not $legacyJson) {
|
||||
if (-not $legacyCommand) {
|
||||
throw "Legacy 4028 test failed: no legacy command in payload."
|
||||
}
|
||||
|
||||
$legacyJson = '{"command":"' + $legacyCommand + '"}'
|
||||
}
|
||||
|
||||
Write-AgentLog "Running legacy 4028 M30 test against $ipAddress command=$legacyCommand"
|
||||
|
||||
$result = Invoke-MinerOpsLegacy4028Command -IpAddress $ipAddress -CommandJson $legacyJson -Port 4028
|
||||
|
||||
return @{
|
||||
success = $true
|
||||
action = "legacy_4028"
|
||||
test_lab = "m30"
|
||||
ip_address = $ipAddress
|
||||
legacy_4028_command = $legacyCommand
|
||||
legacy_4028_request = $legacyJson
|
||||
legacy_4028_response = $result.raw_response
|
||||
legacy_4028_parsed = $result.parsed_response
|
||||
control_method = "legacy_4028"
|
||||
}
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
function Run-MinerControlCommand {
|
||||
@@ -1244,105 +979,7 @@ function Run-MinerControlCommand {
|
||||
[Parameter(Mandatory=$true)]
|
||||
[object]$Payload
|
||||
)
|
||||
# MinerOps M30 legacy 4028 test path
|
||||
if ($Payload -is [string]) {
|
||||
try {
|
||||
$Payload = $Payload | ConvertFrom-Json
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
if ($Payload -and (($Payload.control_method -eq "legacy_4028") -or ($Payload.action -eq "legacy_4028"))) {
|
||||
return Invoke-MinerOpsM30Legacy4028Test -Payload $Payload
|
||||
}
|
||||
|
||||
|
||||
if ($Payload -is [string]) {
|
||||
$Payload = $Payload | ConvertFrom-Json
|
||||
}
|
||||
|
||||
if ($null -eq $Payload) {
|
||||
throw "Miner control payload was empty."
|
||||
}
|
||||
|
||||
$action = ([string]$Payload.action).ToLower()
|
||||
$minerId = [string]$Payload.miner_id
|
||||
$macAddress = [string]$Payload.mac_address
|
||||
$dryRun = [bool]$Payload.dry_run
|
||||
|
||||
$ipAddress = ([string]$Payload.ip_address).Trim()
|
||||
|
||||
if ($ipAddress.Contains("/")) {
|
||||
$ipAddress = $ipAddress.Split("/")[0]
|
||||
}
|
||||
|
||||
if ([string]::IsNullOrWhiteSpace($ipAddress)) {
|
||||
throw "Miner control payload did not include ip_address."
|
||||
}
|
||||
|
||||
$serviceAction = ""
|
||||
$api3Command = ""
|
||||
$api3Param = $null
|
||||
$performanceMode = ""
|
||||
|
||||
if ($action -eq "reboot") {
|
||||
$serviceAction = "restart"
|
||||
$api3Command = "set.miner.service"
|
||||
} elseif ($action -eq "suspend") {
|
||||
$serviceAction = "stop"
|
||||
$api3Command = "set.miner.service"
|
||||
} elseif ($action -eq "resume") {
|
||||
$serviceAction = "start"
|
||||
$api3Command = "set.miner.service"
|
||||
} elseif ($action -eq "performance_mode") {
|
||||
$performanceMode = ([string]$Payload.performance_mode).ToLower()
|
||||
|
||||
if ($performanceMode -in @("low", "normal", "high")) {
|
||||
$api3Command = "set.miner.power_mode"
|
||||
$api3Param = $performanceMode
|
||||
} else {
|
||||
throw "Invalid performance_mode: $performanceMode"
|
||||
}
|
||||
} else {
|
||||
throw "Unsupported miner control action: $action"
|
||||
}
|
||||
|
||||
Write-AgentLog "Miner control requested. Action=$action API3Command=$api3Command API3Service=$serviceAction PerformanceMode=$performanceMode IP=$ipAddress MAC=$macAddress DryRun=$dryRun"
|
||||
|
||||
if ($dryRun) {
|
||||
return @{
|
||||
ok = $true
|
||||
dry_run = $true
|
||||
action = $action
|
||||
api3_command = $api3Command
|
||||
api3_service_action = $serviceAction
|
||||
performance_mode = $performanceMode
|
||||
ip_address = $ipAddress
|
||||
miner_id = $minerId
|
||||
mac_address = $macAddress
|
||||
message = "Dry run only. No miner command sent."
|
||||
}
|
||||
}
|
||||
|
||||
if ($action -eq "performance_mode") {
|
||||
$api3Result = Invoke-MinerOpsApi3AuthenticatedCommand -IpAddress $ipAddress -Command $api3Command -Param $api3Param
|
||||
} else {
|
||||
$api3Result = Invoke-MinerOpsApi3Service -IpAddress $ipAddress -ServiceAction $serviceAction
|
||||
}
|
||||
|
||||
return @{
|
||||
ok = $true
|
||||
dry_run = $false
|
||||
action = $action
|
||||
api3_command = $api3Command
|
||||
api3_service_action = $serviceAction
|
||||
performance_mode = $performanceMode
|
||||
ip_address = $ipAddress
|
||||
miner_id = $minerId
|
||||
mac_address = $macAddress
|
||||
message = "API3 miner control command sent by Windows Agent."
|
||||
api3_result = $api3Result
|
||||
}
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
|
||||
@@ -1368,8 +1005,8 @@ function Process-PendingCommands {
|
||||
Write-AgentLog "Command completed: $commandId"
|
||||
} elseif ($commandType -eq "miner_control") {
|
||||
$result = Run-MinerControlCommand -Payload $cmd.request_payload
|
||||
Send-AgentCommandResult -CommandId $commandId -Status "completed" -ResultPayload $result | Out-Null
|
||||
Write-AgentLog "Miner control command completed: $commandId"
|
||||
Send-AgentCommandResult -CommandId $commandId -Status "failed" -ResultPayload $result | Out-Null
|
||||
Write-AgentLog "Miner control rejected by Stage A containment: $commandId"
|
||||
|
||||
} elseif ($commandType -eq "enrich_match_discovery") {
|
||||
$payloadObj = $cmd.request_payload
|
||||
@@ -1394,7 +1031,12 @@ function Process-PendingCommands {
|
||||
|
||||
Write-AgentLog "Running Enrich / Match for discovery IP $ipAddress"
|
||||
|
||||
$out = powershell.exe -NoProfile -ExecutionPolicy Bypass -File C:\MinerOpsAgent\EnrichOneMiner.ps1 -IpAddress $ipAddress 2>&1 | Out-String
|
||||
$hubConfig = Invoke-AgentApi -Method "GET" -Path "/api/agent/config"
|
||||
if (-not [string]::Equals([string]$payloadObj.agent_zone_id, [string]$hubConfig.agent_zone_id, [StringComparison]::Ordinal)) {
|
||||
throw "discovery_command_stale_authority"
|
||||
}
|
||||
if (-not $payloadObj.discovery_id -or -not $payloadObj.version) { throw "discovery_observation_required" }
|
||||
$out = powershell.exe -NoProfile -ExecutionPolicy Bypass -File C:\MinerOpsAgent\EnrichOneMiner.ps1 -IpAddress $ipAddress -DiscoveryId ([string]$payloadObj.discovery_id) -ObservationVersion ([string]$payloadObj.version) -CommandId $commandId 2>&1 | Out-String
|
||||
|
||||
if ($LASTEXITCODE -ne 0) {
|
||||
throw "EnrichOneMiner failed for $ipAddress. Output: $out"
|
||||
|
||||
@@ -30,11 +30,57 @@ function Invoke-WhatsMinerApi3Read {
|
||||
[int] $TimeoutMs = 2500
|
||||
)
|
||||
|
||||
# Exact operation and parameter allowlist at the shared transport boundary.
|
||||
# No extra keys (including credentials, command aliases, or parameter overrides)
|
||||
# reach the wire, even when this function is called outside the poller loop.
|
||||
$readCommands = @("get.device.info", "get.miner.setting", "get.miner.status",
|
||||
"get.fan.setting", "get.device.custom_data")
|
||||
$command = $Request["cmd"]
|
||||
$approvedCommand = $null
|
||||
if ($command -is [string]) {
|
||||
foreach ($candidate in $readCommands) {
|
||||
if ([string]::Equals($candidate, $command, [StringComparison]::Ordinal)) {
|
||||
$approvedCommand = $candidate
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if ($null -eq $approvedCommand) {
|
||||
throw "unsupported_diagnostic_operation_stage_a"
|
||||
}
|
||||
$expectedKeys = @("cmd")
|
||||
$wireRequest = [ordered]@{ cmd = $approvedCommand }
|
||||
if ([string]::Equals($approvedCommand, "get.miner.status", [StringComparison]::Ordinal)) {
|
||||
$expectedKeys += "param"
|
||||
if ($Request["param"] -isnot [string] -or
|
||||
-not [string]::Equals($Request["param"], "summary+pools+edevs", [StringComparison]::Ordinal)) {
|
||||
throw "unsupported_diagnostic_operation_stage_a"
|
||||
}
|
||||
$wireRequest["param"] = "summary+pools+edevs"
|
||||
}
|
||||
if ($Request.Count -ne $expectedKeys.Count) {
|
||||
throw "unsupported_diagnostic_operation_stage_a"
|
||||
}
|
||||
foreach ($key in $Request.Keys) {
|
||||
$approvedKey = $false
|
||||
if ($key -is [string]) {
|
||||
foreach ($expectedKey in $expectedKeys) {
|
||||
if ([string]::Equals($expectedKey, $key, [StringComparison]::Ordinal)) {
|
||||
$approvedKey = $true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if (-not $approvedKey) {
|
||||
throw "unsupported_diagnostic_operation_stage_a"
|
||||
}
|
||||
}
|
||||
|
||||
$client = $null
|
||||
$stream = $null
|
||||
|
||||
try {
|
||||
$json = ($Request | ConvertTo-Json -Depth 20 -Compress)
|
||||
$json = ($wireRequest | ConvertTo-Json -Depth 20 -Compress)
|
||||
$payload = [System.Text.Encoding]::ASCII.GetBytes($json)
|
||||
$length = [BitConverter]::GetBytes([Int32]$payload.Length)
|
||||
|
||||
@@ -178,7 +224,9 @@ foreach ($miner in $miners) {
|
||||
}
|
||||
|
||||
$items.Add([pscustomobject]@{
|
||||
ip = $ip
|
||||
miner_id = [string]$miner.miner_id
|
||||
miner_version = [string]$miner.miner_version
|
||||
ip = $ip
|
||||
mac = $mac
|
||||
api3_supported = $false
|
||||
api3_error = $err
|
||||
@@ -211,6 +259,8 @@ foreach ($miner in $miners) {
|
||||
-Request @{ cmd = "get.device.custom_data" }
|
||||
|
||||
$items.Add([pscustomobject]@{
|
||||
miner_id = [string]$miner.miner_id
|
||||
miner_version = [string]$miner.miner_version
|
||||
ip = $ip
|
||||
mac = $mac
|
||||
api3_supported = $true
|
||||
@@ -229,15 +279,14 @@ $payload = @{
|
||||
items = @($items.ToArray())
|
||||
}
|
||||
|
||||
$json = $payload | ConvertTo-Json -Depth 80
|
||||
|
||||
Write-Host "Posting API3 enrichment results to hub..."
|
||||
$result = Invoke-RestMethod `
|
||||
-Uri "$HubUrl/api/agent/api3-enrichment" `
|
||||
-Method Post `
|
||||
-Headers $Headers `
|
||||
-ContentType "application/json" `
|
||||
-Body $json
|
||||
$allItems = @($items.ToArray())
|
||||
$result = @()
|
||||
for ($offset = 0; $offset -lt $allItems.Count; $offset += 128) {
|
||||
$last = [Math]::Min($offset + 127, $allItems.Count - 1)
|
||||
$payload.items = @($allItems[$offset..$last])
|
||||
$result += Invoke-RestMethod -Uri "$HubUrl/api/agent/api3-enrichment" -Method Post `
|
||||
-Headers $Headers -ContentType "application/json" -Body ($payload | ConvertTo-Json -Depth 80)
|
||||
}
|
||||
|
||||
Write-Host ""
|
||||
Write-Host "Local checked: $checked"
|
||||
|
||||
@@ -26,29 +26,8 @@ function Get-MinerOpsControlCredentials {
|
||||
}
|
||||
|
||||
function Invoke-MinerOpsControlApi3Command {
|
||||
param(
|
||||
[Parameter(Mandatory=$true)][string]$IpAddress,
|
||||
[Parameter(Mandatory=$true)][string]$Command,
|
||||
[object]$Param = $null
|
||||
)
|
||||
$credentials = Get-MinerOpsControlCredentials
|
||||
if (-not $credentials.available) { throw "credentials_missing" }
|
||||
|
||||
$saltInfo = Get-MinerOpsApi3Salt -IpAddress $IpAddress -Port $credentials.port
|
||||
$timestamp = [int64][DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
|
||||
$token = Get-MinerOpsApi3Token -Command $Command -Password $credentials.password `
|
||||
-Salt ([string]$saltInfo.salt) -Timestamp $timestamp
|
||||
$payload = @{
|
||||
cmd = $Command
|
||||
ts = $timestamp
|
||||
token = $token
|
||||
account = $credentials.account
|
||||
}
|
||||
if ($null -ne $Param) { $payload["param"] = $Param }
|
||||
$response = Invoke-MinerOpsApi3Payload -IpAddress $IpAddress `
|
||||
-Port $credentials.port -Payload $payload
|
||||
if ([int]$response.response.code -ne 0) { throw "command_rejected" }
|
||||
return $response
|
||||
param([string]$IpAddress, [string]$Command, [object]$Param = $null)
|
||||
throw "physical_control_disabled_stage_a"
|
||||
}
|
||||
|
||||
function New-MinerOpsControlResult {
|
||||
@@ -74,90 +53,11 @@ function New-MinerOpsControlResult {
|
||||
|
||||
function Run-MinerControlCommand {
|
||||
param([Parameter(Mandatory=$true)][object]$Payload)
|
||||
|
||||
try {
|
||||
if ($Payload -is [string]) { $Payload = $Payload | ConvertFrom-Json }
|
||||
$action = ([string]$Payload.action).Trim().ToLower()
|
||||
$ipAddress = ([string]$Payload.ip_address).Trim()
|
||||
$requested = ([string]$Payload.requested_value).Trim().ToLower()
|
||||
if (-not $requested -and $Payload.performance_mode) {
|
||||
$requested = ([string]$Payload.performance_mode).Trim().ToLower()
|
||||
}
|
||||
if ($action -notin @("suspend", "resume", "performance_mode")) {
|
||||
return New-MinerOpsControlResult -Accepted $false `
|
||||
-VerificationAttempted $false -VerificationStatus "unavailable" `
|
||||
-FailureCode "invalid_action" -QueueStatus "failed"
|
||||
}
|
||||
if ([string]::IsNullOrWhiteSpace($ipAddress)) {
|
||||
return New-MinerOpsControlResult -Accepted $false `
|
||||
-VerificationAttempted $false -VerificationStatus "unavailable" `
|
||||
-FailureCode "target_missing" -QueueStatus "failed"
|
||||
}
|
||||
$credentials = Get-MinerOpsControlCredentials
|
||||
if (-not $credentials.available) {
|
||||
return New-MinerOpsControlResult -Accepted $false `
|
||||
-VerificationAttempted $false -VerificationStatus "unavailable" `
|
||||
-FailureCode "credentials_missing" -QueueStatus "failed"
|
||||
}
|
||||
|
||||
if ($action -eq "performance_mode") {
|
||||
if ($requested -notin @("low", "normal", "high")) {
|
||||
return New-MinerOpsControlResult -Accepted $false `
|
||||
-VerificationAttempted $false -VerificationStatus "unavailable" `
|
||||
-FailureCode "invalid_mode" -QueueStatus "failed"
|
||||
}
|
||||
Invoke-MinerOpsControlApi3Command -IpAddress $ipAddress `
|
||||
-Command "set.miner.power_mode" -Param $requested | Out-Null
|
||||
|
||||
$observed = ""
|
||||
$readUnavailable = $false
|
||||
for ($attempt = 1; $attempt -le 3; $attempt++) {
|
||||
try {
|
||||
$read = Invoke-MinerOpsControlApi3Command -IpAddress $ipAddress `
|
||||
-Command "get.miner.setting"
|
||||
$candidate = ([string]$read.response.msg.'power-mode').Trim().ToLower()
|
||||
if ($candidate -in @("low", "normal", "high")) {
|
||||
$observed = $candidate
|
||||
if ($observed -eq $requested) {
|
||||
return New-MinerOpsControlResult -Accepted $true `
|
||||
-VerificationAttempted $true -VerificationStatus "confirmed" `
|
||||
-ObservedMode $observed
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
if ($_.Exception.Message -match "unsupported|unknown_command") {
|
||||
$readUnavailable = $true
|
||||
break
|
||||
}
|
||||
}
|
||||
if ($attempt -lt 3) { Start-Sleep -Milliseconds 750 }
|
||||
}
|
||||
if ($readUnavailable) {
|
||||
return New-MinerOpsControlResult -Accepted $true `
|
||||
-VerificationAttempted $true -VerificationStatus "unavailable"
|
||||
}
|
||||
if ($observed) {
|
||||
return New-MinerOpsControlResult -Accepted $true `
|
||||
-VerificationAttempted $true `
|
||||
-VerificationStatus "verification_failed" -ObservedMode $observed
|
||||
}
|
||||
return New-MinerOpsControlResult -Accepted $true `
|
||||
-VerificationAttempted $true -VerificationStatus "unconfirmed"
|
||||
}
|
||||
|
||||
$serviceAction = if ($action -eq "suspend") { "stop" } else { "start" }
|
||||
Invoke-MinerOpsControlApi3Command -IpAddress $ipAddress `
|
||||
-Command "set.miner.service" -Param $serviceAction | Out-Null
|
||||
return New-MinerOpsControlResult -Accepted $true `
|
||||
-VerificationAttempted $false -VerificationStatus "unconfirmed"
|
||||
} catch {
|
||||
$failure = if ($_.Exception.Message -eq "credentials_missing") {
|
||||
"credentials_missing"
|
||||
} else { "write_rejected" }
|
||||
return New-MinerOpsControlResult -Accepted $false `
|
||||
-VerificationAttempted $false -VerificationStatus "unavailable" `
|
||||
-FailureCode $failure -QueueStatus "failed"
|
||||
}
|
||||
# No current request schema binds authorization to physical identity.
|
||||
# Do not parse, normalize, reinterpret, or transmit an old queued payload.
|
||||
return New-MinerOpsControlResult -Accepted $false `
|
||||
-VerificationAttempted $false -VerificationStatus "unavailable" `
|
||||
-FailureCode "physical_control_disabled_stage_a" -QueueStatus "failed"
|
||||
}
|
||||
|
||||
function Send-AgentCommandResult {
|
||||
@@ -185,7 +85,6 @@ function Send-AgentCommandResult {
|
||||
|
||||
function Send-AgentCheckin {
|
||||
$config = Read-AgentConfig
|
||||
$credentials = Get-MinerOpsControlCredentials
|
||||
$body = [ordered]@{
|
||||
agent_id = [string]$config.agent_id
|
||||
site_id = [string]$config.site_id
|
||||
@@ -193,8 +92,8 @@ function Send-AgentCheckin {
|
||||
hostname = $env:COMPUTERNAME
|
||||
version = Get-MinerOpsAgentVersion
|
||||
ip_address = ""
|
||||
control_available = [bool]$credentials.available
|
||||
control_unavailable_reason = if ($credentials.available) { $null } else { "credentials_missing" }
|
||||
control_available = $false
|
||||
control_unavailable_reason = "physical_control_disabled_stage_a"
|
||||
local_time = (Get-Date).ToString("o")
|
||||
}
|
||||
return Invoke-AgentApi -Method "POST" -Path "/api/agent/checkin" -Body $body
|
||||
|
||||
@@ -120,6 +120,7 @@ foreach ($miner in $knownMiners) {
|
||||
if ($pools.Count -gt 0) {
|
||||
$poolSnapshots.Add([pscustomobject]@{
|
||||
miner_id = $miner.miner_id
|
||||
miner_version = $miner.miner_version
|
||||
ip_address = $ip
|
||||
mac_address = $miner.mac_address
|
||||
model = $miner.model
|
||||
@@ -130,18 +131,14 @@ foreach ($miner in $knownMiners) {
|
||||
|
||||
Write-Host "Pool snapshots collected: $($poolSnapshots.Count)"
|
||||
|
||||
$body = @{
|
||||
agent_id = $AgentId
|
||||
pool_snapshots = $poolSnapshots
|
||||
} | ConvertTo-Json -Depth 20
|
||||
|
||||
$result = Invoke-RestMethod `
|
||||
-Method Post `
|
||||
-Uri "$HubUrl/api/agent/pool-snapshots" `
|
||||
-Headers $Headers `
|
||||
-Body $body `
|
||||
-ContentType "application/json" `
|
||||
-TimeoutSec 180
|
||||
$allItems = @($poolSnapshots.ToArray())
|
||||
$result = @()
|
||||
for ($offset = 0; $offset -lt $allItems.Count; $offset += 128) {
|
||||
$last = [Math]::Min($offset + 127, $allItems.Count - 1)
|
||||
$body = @{agent_id=$AgentId; pool_snapshots=@($allItems[$offset..$last])} | ConvertTo-Json -Depth 80
|
||||
$result += Invoke-RestMethod -Method Post -Uri "$HubUrl/api/agent/pool-snapshots" `
|
||||
-Headers $Headers -Body $body -ContentType "application/json" -TimeoutSec 180
|
||||
}
|
||||
|
||||
$result | ConvertTo-Json -Depth 10
|
||||
|
||||
@@ -1,7 +1,44 @@
|
||||
# MinerOps Windows Agent
|
||||
|
||||
## Stage A physical-control containment
|
||||
|
||||
Physical miner control is disabled in this version, including suspend, resume,
|
||||
performance-mode changes, reboot, M30 tests, and miner-side password changes.
|
||||
No environment flag, configured API3 credential, or Agent capability report can
|
||||
enable it. This is containment, not completion of Blocker #3 identity binding.
|
||||
|
||||
The Hub returns HTTP 410 for retired legacy command, scheduling, and unrestricted
|
||||
diagnostic endpoints (after authorization where applicable). Typed control POSTs
|
||||
require `miners.control` and return HTTP 409 `physical_control_disabled_stage_a`.
|
||||
The unrestricted `whatsminer-read-test` endpoint is not a supported read-only tool.
|
||||
|
||||
Authenticated queue polling cancels pending/claimed physical and unknown command
|
||||
types, records an atomic `agent.command.rejected` audit entry, and only delivers
|
||||
fixed telemetry/discovery command types. Results cannot promote retired entries.
|
||||
Original payloads, prior results/errors, timestamps, and terminal history remain.
|
||||
Cancellation of an already claimed command does **not** prove it was never sent;
|
||||
the audit explicitly records its previous execution outcome as unknown.
|
||||
|
||||
The Windows executor rejects all current physical-control schemas without any
|
||||
device connection and advertises physical control as unavailable. Existing Agents
|
||||
must be stopped/drained and upgraded before claiming fleet-wide containment:
|
||||
server cancellation cannot recall payloads cached by older running Agents.
|
||||
Do not enable control or deploy this work before separate security review.
|
||||
The ordered, fail-closed procedure is in
|
||||
[the Stage A old-Agent cutover runbook](../../docs/operations/STAGE_A_OLD_AGENT_CUTOVER.txt).
|
||||
Native Windows PowerShell 5.1/SYSTEM validation and verified fleet network fencing
|
||||
remain production gates; server cancellation alone cannot contain an old executor.
|
||||
|
||||
Retained legacy/API3 enrichment helpers validate exact read commands and fixed
|
||||
parameters before opening a socket. Unknown operations and extra keys fail closed.
|
||||
Schedule cancel/delete require `miners.control` and retire matching live unsupported
|
||||
commands for the stored Agent with atomic actor-attributed rejection audits.
|
||||
Deletion retains schedule history; retries do not duplicate audits. Previously
|
||||
claimed execution remains unknown. Manual cutover retirement does not claim read
|
||||
work and must sweep offline/queue-only Agent IDs until the global live count is zero.
|
||||
|
||||
The Windows Agent runs at a mining site and performs local discovery, polling,
|
||||
API3 enrichment, pool polling, and approved queued miner-control operations.
|
||||
API3 enrichment, and pool polling. Physical miner control is disabled in Stage A.
|
||||
MinerOps does not send API3 credentials through the hub command queue; those
|
||||
credentials remain in the Agent's local `agent-config.json`.
|
||||
|
||||
|
||||
@@ -1,84 +1,115 @@
|
||||
$ErrorActionPreference = "Stop"
|
||||
$script:config = [pscustomobject]@{ api3_account = "operator"; api3_password = "secret"; api3_port = 4433 }
|
||||
$script:calls = @()
|
||||
$script:outcomes = @()
|
||||
$script:outcomeIndex = 0
|
||||
|
||||
function Read-AgentConfig { return $script:config }
|
||||
. (Join-Path (Split-Path $PSScriptRoot -Parent) "MinerOpsControl.ps1")
|
||||
|
||||
function Invoke-MinerOpsControlApi3Command {
|
||||
param([string]$IpAddress, [string]$Command, [object]$Param = $null)
|
||||
$script:calls += [pscustomobject]@{ command = $Command; param = $Param }
|
||||
$outcome = $script:outcomes[$script:outcomeIndex]
|
||||
$script:outcomeIndex += 1
|
||||
if ($outcome -is [System.Exception]) { throw $outcome }
|
||||
return $outcome
|
||||
}
|
||||
$script:assertions = 0
|
||||
$script:physicalCalls = 0
|
||||
$script:readCalls = 0
|
||||
$script:apiCalls = @()
|
||||
$script:commands = @()
|
||||
|
||||
function Assert-Equal($Actual, $Expected, [string]$Message) {
|
||||
$script:assertions++
|
||||
if ($Actual -ne $Expected) { throw "$Message. Expected '$Expected', got '$Actual'." }
|
||||
}
|
||||
function Reset-Harness([object[]]$Outcomes) {
|
||||
$script:calls = @()
|
||||
$script:outcomes = @($Outcomes)
|
||||
$script:outcomeIndex = 0
|
||||
}
|
||||
function Response([string]$Mode = "") {
|
||||
return @{ response = @{ code = 0; msg = @{ "power-mode" = $Mode } } }
|
||||
}
|
||||
|
||||
$payload = @{ action = "performance_mode"; requested_value = "normal"; ip_address = "192.0.2.1" }
|
||||
|
||||
$script:config = [pscustomobject]@{ api3_account = ""; api3_password = ""; api3_port = 4433 }
|
||||
Reset-Harness @()
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
Assert-Equal $result.failure_code "credentials_missing" "Missing credentials code"
|
||||
Assert-Equal $script:calls.Count 0 "Missing credentials must not contact API3"
|
||||
|
||||
$script:config = [pscustomobject]@{ api3_account = "operator"; api3_password = "secret"; api3_port = 4433 }
|
||||
Reset-Harness @((Response), (Response "normal"))
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
Assert-Equal $result.verification_status "confirmed" "Performance confirmation"
|
||||
Assert-Equal @($script:calls | Where-Object command -Like "set.*").Count 1 "Exactly one write"
|
||||
|
||||
Reset-Harness @((Response), (Response "low"), (Response "low"), (Response "low"))
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
Assert-Equal $result.verification_status "verification_failed" "Contradictory mode"
|
||||
Assert-Equal $script:calls.Count 4 "One write and three bounded reads"
|
||||
|
||||
Reset-Harness @((Response), (Response), (Response), (Response))
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
Assert-Equal $result.verification_status "unconfirmed" "Ambiguous reads"
|
||||
|
||||
Reset-Harness @((Response), [System.Exception]::new("unknown_command"))
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
Assert-Equal $result.verification_status "unavailable" "Unavailable read path"
|
||||
|
||||
foreach ($action in @("suspend", "resume")) {
|
||||
Reset-Harness @((Response))
|
||||
$result = Run-MinerControlCommand -Payload @{
|
||||
action = $action; requested_value = $null; ip_address = "192.0.2.1"
|
||||
function Assert-Throws($Block, [string]$Message) {
|
||||
$caught = $false
|
||||
try { & $Block | Out-Null } catch {
|
||||
$caught = $true
|
||||
Assert-Equal $_.Exception.Message "physical_control_disabled_stage_a" $Message
|
||||
}
|
||||
Assert-Equal $result.command_accepted $true "$action accepted"
|
||||
Assert-Equal $result.verification_status "unconfirmed" "$action remains unconfirmed"
|
||||
Assert-Equal $result.observed_service_state $null "$action has no invented state"
|
||||
Assert-Equal @($script:calls | Where-Object command -Like "set.*").Count 1 "$action one write"
|
||||
Assert-Equal $caught $true "$Message must reject"
|
||||
}
|
||||
|
||||
Reset-Harness @([System.Exception]::new("vendor response included secret"))
|
||||
$result = Run-MinerControlCommand -Payload @{
|
||||
action = "suspend"; requested_value = $null; ip_address = "192.0.2.1"
|
||||
# Define functions from the real service AST, without executing its service loop,
|
||||
# config loading, startup, or any scheduled tasks.
|
||||
$root = Split-Path $PSScriptRoot -Parent
|
||||
$tokens = $null; $parseErrors = $null
|
||||
$ast = [System.Management.Automation.Language.Parser]::ParseFile(
|
||||
(Join-Path $root "MinerOpsAgentService.ps1"), [ref]$tokens, [ref]$parseErrors)
|
||||
Assert-Equal $parseErrors.Count 0 "Service parser"
|
||||
foreach ($function in $ast.FindAll({
|
||||
param($node)
|
||||
$node -is [System.Management.Automation.Language.FunctionDefinitionAst] -and
|
||||
$node.Parent -is [System.Management.Automation.Language.NamedBlockAst]
|
||||
}, $true)) {
|
||||
Invoke-Expression $function.Extent.Text
|
||||
}
|
||||
Assert-Equal $result.failure_code "write_rejected" "Write rejection is sanitized"
|
||||
$json = $result | ConvertTo-Json -Compress
|
||||
if ($json -match "vendor|secret|192\\.0\\.2\\.1") { throw "Typed result leaked raw data" }
|
||||
|
||||
Reset-Harness @()
|
||||
$result = Run-MinerControlCommand -Payload @{
|
||||
action = "performance_mode"; requested_value = "turbo"; ip_address = "192.0.2.1"
|
||||
$script:productionApi3Payload = (Get-Command Invoke-MinerOpsApi3Payload).ScriptBlock
|
||||
$script:productionPlainCommand = (Get-Command Invoke-MinerPlainCommand).ScriptBlock
|
||||
function Invoke-MinerOpsApi3Payload { $script:physicalCalls++; throw "Unexpected API3 transport" }
|
||||
function Get-MinerOpsApi3Salt { $script:physicalCalls++; throw "Unexpected salt probe" }
|
||||
function Invoke-MinerPlainCommand { $script:physicalCalls++; throw "Unexpected legacy transport" }
|
||||
function Read-AgentConfig { return [pscustomobject]@{
|
||||
agent_id = "simulated-agent"; site_id = ""; site_name = "Simulation"
|
||||
api3_account = "operator"; api3_password = "mock-only"; api3_port = 4433
|
||||
} }
|
||||
function Write-AgentLog { param([string]$Message) }
|
||||
function Invoke-AgentApi {
|
||||
param([string]$Method, [string]$Path, [object]$Body)
|
||||
$script:apiCalls += @{ method = $Method; path = $Path; body = $Body }
|
||||
return @{ ok = $true }
|
||||
}
|
||||
Assert-Equal $result.failure_code "invalid_mode" "Invalid mode code"
|
||||
Assert-Equal $script:calls.Count 0 "Invalid mode must not contact API3"
|
||||
function Get-PendingAgentCommands { return $script:commands }
|
||||
function Run-KnownMinerPoll { $script:readCalls++; return @{ simulated = $true } }
|
||||
function Run-NewMinerDiscovery { $script:readCalls++; return @{ simulated = $true } }
|
||||
|
||||
Write-Host "MinerOps typed Agent control tests passed."
|
||||
# The pre-module legacy implementation must also be inert, including helpers
|
||||
# which used to support direct generic API3 or M30/4028 writes.
|
||||
Assert-Throws { Run-MinerControlCommand -Payload @{ action = "reboot"; dry_run = $false } } "Legacy executor"
|
||||
Assert-Throws { Invoke-MinerOpsApi3Service -IpAddress "192.0.2.1" -ServiceAction "stop" } "Legacy service helper"
|
||||
Assert-Throws { & $script:productionApi3Payload -IpAddress "192.0.2.1" -Payload @{ cmd = "set.miner.service"; param = "stop" } } "Generic API3 payload helper"
|
||||
Assert-Throws { & $script:productionPlainCommand -IpAddress "192.0.2.1" -Command "reboot" } "Generic legacy command helper"
|
||||
Assert-Throws { Invoke-MinerOpsApi3AuthenticatedCommand -IpAddress "192.0.2.1" -Command "set.miner.service" -Param "stop" } "Legacy API3"
|
||||
Assert-Throws { Invoke-MinerOpsLegacy4028Command -IpAddress "192.0.2.1" -CommandJson '{"command":"reboot"}' } "Legacy 4028"
|
||||
Assert-Throws { Invoke-MinerOpsM30Legacy4028Test -Payload @{ legacy_4028_command = "reboot" } } "M30 helper"
|
||||
|
||||
. (Join-Path $root "MinerOpsControl.ps1")
|
||||
Assert-Throws { Invoke-MinerOpsControlApi3Command -IpAddress "192.0.2.1" -Command "set.miner.service" -Param "stop" } "Typed transport"
|
||||
Assert-Throws { Invoke-MinerOpsControlApi3Command -IpAddress "192.0.2.1" -Command "get.read_then_write" } "Disguised diagnostic"
|
||||
|
||||
$payloads = @(
|
||||
@{ action = "legacy_4028"; legacy_4028_command = "reboot"; dry_run = $false },
|
||||
@{ action = "reboot"; ip_address = "192.0.2.1" },
|
||||
@{ schema_version = 1; action = "suspend"; ip_address = "192.0.2.1"; miner_id = "spoofed" },
|
||||
@{ schema_version = 1; action = "resume"; ip_address = "192.0.2.1" },
|
||||
@{ schema_version = 1; action = "performance_mode"; requested_value = "high"; ip_address = "192.0.2.1" },
|
||||
@{ schema_version = 999; action = "suspend" },
|
||||
@{ schema_version = "1"; action = "performance_mode"; performance_mode = "high" },
|
||||
@{ command = "summary"; params = @{ cmd = "set.miner.service"; param = "stop" } },
|
||||
@{ action = "read_reboot"; dry_run = $true },
|
||||
@{},
|
||||
'{"schema_version":0,"action":"reboot"}',
|
||||
"{invalid-json"
|
||||
)
|
||||
foreach ($payload in $payloads) {
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
Assert-Equal $result.command_accepted $false "Rejected payload acceptance"
|
||||
Assert-Equal $result.failure_code "physical_control_disabled_stage_a" "Rejected payload reason"
|
||||
Assert-Equal $result._queue_status "failed" "Rejected payload terminal result"
|
||||
Assert-Equal $result.verification_attempted $false "No physical verification/probe"
|
||||
Assert-Equal $result.observed_service_state $null "No invented device state"
|
||||
Assert-Equal $script:physicalCalls 0 "No physical calls"
|
||||
$script:commands = @(@{ command_id = "legacy-delivery"; command_type = "miner_control"; request_payload = $payload })
|
||||
$script:apiCalls = @()
|
||||
Assert-Equal (Process-PendingCommands) 1 "Delivered legacy command handled"
|
||||
Assert-Equal $script:apiCalls.Count 1 "One failed result submitted"
|
||||
Assert-Equal $script:apiCalls[0].body.status "failed" "Legacy delivery failed"
|
||||
Assert-Equal $script:apiCalls[0].body.result_payload.command_accepted $false "Legacy delivery not accepted"
|
||||
Assert-Equal $script:physicalCalls 0 "No writes after delivery"
|
||||
}
|
||||
foreach ($kind in @("raw_api3", "miner_control_v2", "read_reboot", "")) {
|
||||
$script:commands = @(@{ command_id = "unsupported"; command_type = $kind; request_payload = @{ action = "reboot" } })
|
||||
$script:apiCalls = @()
|
||||
Assert-Equal (Process-PendingCommands) 1 "Unknown command handled"
|
||||
Assert-Equal $script:apiCalls[0].body.status "failed" "Unknown command fails closed"
|
||||
}
|
||||
Assert-Equal $script:readCalls 0 "No physical request reinterpreted as telemetry"
|
||||
$script:commands = @(@{ command_id = "read-only"; command_type = "poll_known_miners_now"; request_payload = @{} })
|
||||
$script:apiCalls = @()
|
||||
Assert-Equal (Process-PendingCommands) 1 "Fixed read-only command available"
|
||||
Assert-Equal $script:readCalls 1 "Only simulated read performed"
|
||||
Assert-Equal $script:physicalCalls 0 "No physical calls in entire suite"
|
||||
$script:apiCalls = @()
|
||||
Send-AgentCheckin | Out-Null
|
||||
Assert-Equal $script:apiCalls[0].body.control_available $false "Cannot advertise physical capability"
|
||||
Assert-Equal $script:apiCalls[0].body.control_unavailable_reason "physical_control_disabled_stage_a" "Containment advertised"
|
||||
Write-Host "Stage A Windows containment passed: $($script:assertions) assertions; zero physical transport calls."
|
||||
@@ -0,0 +1,199 @@
|
||||
$ErrorActionPreference = "Stop"
|
||||
$script:assertions = 0
|
||||
$script:connections = 0
|
||||
$TimeoutMs = 1000
|
||||
|
||||
function Assert-True([bool]$Condition, [string]$Message) {
|
||||
$script:assertions++
|
||||
if (-not $Condition) { throw $Message }
|
||||
}
|
||||
function Assert-Rejected([scriptblock]$Call) {
|
||||
$before = $script:connections
|
||||
$caught = $false
|
||||
try { & $Call | Out-Null } catch {
|
||||
$caught = $true
|
||||
Assert-True ($_.Exception.Message -eq "unsupported_diagnostic_operation_stage_a") "Must reject before transport"
|
||||
}
|
||||
Assert-True $caught "Unsupported diagnostics must throw"
|
||||
Assert-True ($script:connections -eq $before) "Rejected request must create zero sockets and write zero bytes"
|
||||
}
|
||||
function Import-Definitions([string]$Path) {
|
||||
$tokens = $null; $errors = $null
|
||||
$ast = [System.Management.Automation.Language.Parser]::ParseFile($Path, [ref]$tokens, [ref]$errors)
|
||||
Assert-True ($errors.Count -eq 0) "Production script must parse"
|
||||
foreach ($definition in $ast.FindAll({
|
||||
param($node)
|
||||
$node -is [System.Management.Automation.Language.FunctionDefinitionAst] -and
|
||||
$node.Parent -is [System.Management.Automation.Language.NamedBlockAst]
|
||||
}, $true)) {
|
||||
. ([scriptblock]::Create("function global:" + $definition.Name + " " + $definition.Body.Extent.Text))
|
||||
}
|
||||
}
|
||||
|
||||
# Load only real function definitions: no config, Hub request, or scheduled loop.
|
||||
$root = Split-Path $PSScriptRoot -Parent
|
||||
Import-Definitions (Join-Path $root "EnrichOneMiner.ps1")
|
||||
Import-Definitions (Join-Path $root "MinerOpsApi3EnrichmentPoller.ps1")
|
||||
Import-Definitions (Join-Path $root "MinerOpsPoolPoller.ps1")
|
||||
|
||||
function New-Object {
|
||||
param([Parameter(Position=0)][string]$TypeName,
|
||||
[Parameter(Position=1)][object[]]$ArgumentList)
|
||||
if ($TypeName -eq "System.Net.Sockets.TcpClient") { $script:connections++ }
|
||||
if ($null -ne $ArgumentList) {
|
||||
return Microsoft.PowerShell.Utility\New-Object -TypeName $TypeName -ArgumentList $ArgumentList
|
||||
}
|
||||
return Microsoft.PowerShell.Utility\New-Object -TypeName $TypeName
|
||||
}
|
||||
|
||||
# Auditor reproductions plus write families, alias/parameter overrides, injection,
|
||||
# malformed structures and operations that merely contain a read-looking substring.
|
||||
foreach ($command in @("reboot", "set.miner.service", "power_off", "switchpool", "ascset",
|
||||
"summary|reboot", "summary`nreboot", 'summary","cmd":"reboot', "read_then_reboot",
|
||||
"SUMMARY", "", "get.device.info", "get_miner_info|all")) {
|
||||
Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $command }
|
||||
}
|
||||
Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command "summary" -Port 4433 }
|
||||
$badRequests = @(
|
||||
@{ cmd = "set.miner.service"; param = "stop" },
|
||||
@{ cmd = "reboot" }, @{ cmd = "set.miner.power_mode"; param = "high" },
|
||||
@{ cmd = "set.device.password"; param = "changed" }, @{ cmd = "get.read_then_write" },
|
||||
@{ cmd = "get.device.info"; command = "reboot" },
|
||||
@{ cmd = "get.device.info"; param = "stop" },
|
||||
@{ cmd = "get.device.info"; account = "operator"; token = "anything" },
|
||||
@{ cmd = "get.miner.status" }, @{ cmd = "get.miner.status"; param = "all" },
|
||||
@{ cmd = "get.miner.status"; param = @("summary+pools+edevs") },
|
||||
@{ cmd = "get.miner.status"; param = "summary+pools+edevs"; command = "reboot" },
|
||||
@{ Cmd = "get.device.info" }, @{ cmd = "GET.DEVICE.INFO" },
|
||||
@{ cmd = "get.device.info`nreboot" }, @{ cmd = @("get.device.info", "reboot") },
|
||||
@{ cmd = $null }, @{}
|
||||
)
|
||||
foreach ($request in $badRequests) {
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request $request }
|
||||
Assert-Rejected { Invoke-Api3OkCommand -Ip "127.0.0.1" -Request $request }
|
||||
}
|
||||
# Exact re-audit suffixes: U+0000, U+00AD and U+200B; also prefix/embedded
|
||||
# invisible characters, controls and whitespace. Never trim or normalize inputs.
|
||||
$originalCulture = [Threading.Thread]::CurrentThread.CurrentCulture
|
||||
try {
|
||||
foreach ($cultureName in @("en-US", "tr-TR", "")) {
|
||||
[Threading.Thread]::CurrentThread.CurrentCulture = [Globalization.CultureInfo]::GetCultureInfo($cultureName)
|
||||
$variants = @()
|
||||
foreach ($code in @(0, 0x00ad, 0x200b, 0x200c, 0x200d, 0x2060, 0xfeff,
|
||||
0x001f, 13, 10, 9, 32, 0x00a0)) {
|
||||
$character = [string][char]$code
|
||||
foreach ($position in @("prefix", "embedded", "suffix")) {
|
||||
$variants += @{ Character = $character; Position = $position }
|
||||
}
|
||||
}
|
||||
foreach ($variant in $variants) {
|
||||
$legacy = "summary"; $api = "get.device.info"; $key = "cmd"; $param = "summary+pools+edevs"
|
||||
if ($variant.Position -eq "prefix") {
|
||||
$legacy = $variant.Character + $legacy; $api = $variant.Character + $api
|
||||
$key = $variant.Character + $key; $param = $variant.Character + $param
|
||||
} elseif ($variant.Position -eq "embedded") {
|
||||
$legacy = "sum" + $variant.Character + "mary"; $api = "get.device." + $variant.Character + "info"
|
||||
$key = "c" + $variant.Character + "md"; $param = "summary+" + $variant.Character + "pools+edevs"
|
||||
} else {
|
||||
$legacy += $variant.Character; $api += $variant.Character
|
||||
$key += $variant.Character; $param += $variant.Character
|
||||
}
|
||||
Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $legacy }
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{ cmd = $api } }
|
||||
Assert-Rejected { Invoke-Api3OkCommand -Ip "127.0.0.1" -Request @{ cmd = $api } }
|
||||
$request = @{}; $request[$key] = "get.device.info"
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request $request }
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{ cmd = "get.miner.status"; param = $param } }
|
||||
$request = @{ cmd = "get.miner.status" }; $request[$key.Replace("cmd", "param")] = "summary+pools+edevs"
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request $request }
|
||||
}
|
||||
foreach ($legacy in @(("summar" + [char]0x0443), ([char]0xff53 + "ummary"),
|
||||
("summ" + [char]0x0061 + [char]0x0301 + "ry"),
|
||||
("summ" + [char]0x00e1 + "ry"), "Summary", "summary/", "summary+")) {
|
||||
Assert-Rejected { Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $legacy }
|
||||
}
|
||||
foreach ($api in @(("get.dev" + [char]0x0456 + "ce.info"), ("get" + [char]0xff0e + "device.info"),
|
||||
([char]0xff47 + "et.device.info"), "Get.Device.Info", "get.device.info/", "get.device.info+")) {
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{cmd=$api} }
|
||||
}
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{cmd="get.miner.status"; Param="summary+pools+edevs"} }
|
||||
Assert-Rejected { Invoke-WhatsMinerApi3Read -Ip "127.0.0.1" -Request @{cmd="get.miner.status"; param="Summary+pools+edevs"} }
|
||||
}
|
||||
} finally { [Threading.Thread]::CurrentThread.CurrentCulture = $originalCulture }
|
||||
Assert-True ($script:connections -eq 0) "All negative cases produced zero transport connections/writes"
|
||||
|
||||
# Real TCP compatibility tests: bind loopback only, one request per simulator.
|
||||
# This suite must be run with physical networks disconnected (Docker --network none).
|
||||
Add-Type -TypeDefinition @'
|
||||
using System;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
public class MinerOpsDiagnosticSimulator : IDisposable {
|
||||
TcpListener listener;
|
||||
Thread thread;
|
||||
public string Payload;
|
||||
public string Error;
|
||||
public MinerOpsDiagnosticSimulator(int port, bool framed) {
|
||||
listener = new TcpListener(IPAddress.Loopback, port); listener.Start();
|
||||
thread = new Thread(() => {
|
||||
try {
|
||||
using (var client = listener.AcceptTcpClient()) {
|
||||
client.ReceiveTimeout = 2000;
|
||||
var stream = client.GetStream(); var buffer = new byte[65536]; int count;
|
||||
if (framed) {
|
||||
var length = new byte[4]; ReadExact(stream, length, 4);
|
||||
count = BitConverter.ToInt32(length, 0);
|
||||
if(count < 1 || count > buffer.Length) throw new Exception("Bad request length");
|
||||
ReadExact(stream, buffer, count);
|
||||
} else { count = stream.Read(buffer, 0, buffer.Length); }
|
||||
Payload = Encoding.ASCII.GetString(buffer, 0, count);
|
||||
var response = Encoding.ASCII.GetBytes("{\"code\":0,\"POOLS\":[],\"msg\":{}}");
|
||||
if(framed) { var length = BitConverter.GetBytes(response.Length); stream.Write(length,0,4); }
|
||||
stream.Write(response, 0, response.Length);
|
||||
}
|
||||
} catch(Exception e) { Error = e.Message; }
|
||||
finally { listener.Stop(); }
|
||||
}); thread.IsBackground = true; thread.Start();
|
||||
}
|
||||
static void ReadExact(NetworkStream stream, byte[] bytes, int count) {
|
||||
int offset=0;
|
||||
while(offset<count) { int n=stream.Read(bytes,offset,count-offset); if(n==0) throw new Exception("EOF"); offset+=n; }
|
||||
}
|
||||
public void Dispose() { listener.Stop(); thread.Join(3000); }
|
||||
}
|
||||
'@
|
||||
|
||||
foreach ($command in @("summary", "devs", "edevs", "stats", "pools", "devdetails", "version", "get_version", "get_miner_info")) {
|
||||
$miner = [MinerOpsDiagnosticSimulator]::new(4028, $false)
|
||||
try {
|
||||
$null = Invoke-MinerCommandRaw -IpAddress "127.0.0.1" -Command $command -TimeoutMs 100
|
||||
$observed = $miner.Payload | ConvertFrom-Json
|
||||
Assert-True ([string]::Equals($observed.cmd, $command, [StringComparison]::Ordinal)) "Legacy query must remain compatible"
|
||||
Assert-True ($observed.PSObject.Properties.Count -eq 1) "Legacy request contains only selected read"
|
||||
Assert-True (-not $miner.Error) "Simulator received a valid legacy request"
|
||||
} finally { $miner.Dispose() }
|
||||
}
|
||||
foreach ($command in @("get.device.info", "get.miner.setting", "get.miner.status", "get.fan.setting", "get.device.custom_data")) {
|
||||
$request = @{ cmd = $command }
|
||||
if ($command -ceq "get.miner.status") { $request.param = "summary+pools+edevs" }
|
||||
$miner = [MinerOpsDiagnosticSimulator]::new(4433, $true)
|
||||
try {
|
||||
$result = Invoke-Api3OkCommand -Ip "127.0.0.1" -Request $request
|
||||
$observed = $miner.Payload | ConvertFrom-Json
|
||||
Assert-True ($null -ne $result) "Existing API3 reader wrapper remains compatible"
|
||||
Assert-True ([string]::Equals($observed.cmd, $command, [StringComparison]::Ordinal)) "Exact API3 query transmitted"
|
||||
Assert-True (@($observed.PSObject.Properties).Count -eq $request.Count) "No additional API3 request keys"
|
||||
if ($command -ceq "get.miner.status") {
|
||||
Assert-True ([string]::Equals($observed.param, "summary+pools+edevs", [StringComparison]::Ordinal)) "Only exact read parameter transmitted"
|
||||
}
|
||||
Assert-True (-not $miner.Error) "Simulator received a valid framed read"
|
||||
} finally { $miner.Dispose() }
|
||||
}
|
||||
$miner = [MinerOpsDiagnosticSimulator]::new(4028, $false)
|
||||
try {
|
||||
$null = Invoke-MinerOps4028PoolsSimple -Ip "127.0.0.1"
|
||||
Assert-True (($miner.Payload | ConvertFrom-Json).command -ceq "pools") "Pool poller sends a constant read"
|
||||
} finally { $miner.Dispose() }
|
||||
Write-Host "Diagnostic containment passed: $($script:assertions) assertions; all rejected operations made zero connections/writes."
|
||||
@@ -0,0 +1,55 @@
|
||||
param([string]$AgentRoot = (Join-Path $PSScriptRoot '..'))
|
||||
$ErrorActionPreference = 'Stop'
|
||||
$checks=0
|
||||
foreach ($file in @('MinerOpsAgentService.ps1','EnrichOneMiner.ps1','MinerOpsApi3EnrichmentPoller.ps1','MinerOpsPoolPoller.ps1')) {
|
||||
$tokens=$null; $errors=$null
|
||||
$ast=[System.Management.Automation.Language.Parser]::ParseFile((Join-Path $AgentRoot $file),[ref]$tokens,[ref]$errors)
|
||||
if ($errors.Count) { throw "Parse errors in ${file}: $errors" }
|
||||
$checks++
|
||||
if ($file -eq 'MinerOpsAgentService.ps1') {
|
||||
$function=$ast.Find({param($n) $n -is [System.Management.Automation.Language.FunctionDefinitionAst] -and $n.Name -eq 'Invoke-AgentApi'},$true)
|
||||
Invoke-Expression $function.Extent.Text
|
||||
}
|
||||
}
|
||||
function Read-AgentConfig { return @{hub_url='https://hub.invalid';agent_id='simulated-agent';agent_secret='disposable-only'} }
|
||||
$script:requests=@()
|
||||
function Invoke-RestMethod {
|
||||
param($Method,$Uri,$Headers,$ContentType,$Body,$TimeoutSec)
|
||||
$parsed=$Body | ConvertFrom-Json
|
||||
$script:requests+=@{body=$parsed;headers=$Headers}
|
||||
return @{ok=$true}
|
||||
}
|
||||
foreach ($selection in @(@{path='/api/agent/discoveries';field='discoveries'},@{path='/api/agent/miner-snapshots';field='miners'})) {
|
||||
$script:requests=@();$body=@{agent_id='simulated-agent'}
|
||||
$body[$selection.field]=@(1..257 | ForEach-Object { @{sample=$_;miner_version='signed-version'} })
|
||||
$result=Invoke-AgentApi -Method POST -Path $selection.path -Body $body
|
||||
$sizes=@($script:requests | ForEach-Object { @($_.body.($selection.field)).Count })
|
||||
if (($sizes -join ',') -ne '128,128,1') { throw "Unsafe batching: $sizes" }
|
||||
foreach ($request in $script:requests) {
|
||||
if ($request.headers['X-MinerOps-Agent-ID'] -ne 'simulated-agent') { throw 'Identity lost' }
|
||||
}
|
||||
if ($result.processed -ne 257) { throw 'Samples lost' }
|
||||
$checks+=3
|
||||
}
|
||||
# Execute the actual target-validation prefix only; transport functions/loops are never loaded.
|
||||
$source=Get-Content -LiteralPath (Join-Path $AgentRoot 'EnrichOneMiner.ps1') -Raw
|
||||
$start=$source.IndexOf('$config = Read-AgentConfig')
|
||||
$finish=$source.IndexOf('$commands = @(',$start)
|
||||
$prefix=$source.Substring($start,$finish-$start)
|
||||
$script:targetRequests=0
|
||||
function Invoke-RestMethod {
|
||||
param($Method,$Uri,$Headers,$ContentType,$Body)
|
||||
$script:targetRequests++
|
||||
if ($Uri -notlike '*/api/agent/discovery-target') { throw 'Unexpected request' }
|
||||
return @{ok=$true;ip_address='198.51.100.10'}
|
||||
}
|
||||
$DiscoveryId='';$ObservationVersion='';$CommandId='';$IpAddress='203.0.113.99'
|
||||
$denied=$false
|
||||
try { Invoke-Expression $prefix } catch { $denied=$_.Exception.Message -eq 'discovery_observation_required' }
|
||||
if (!$denied -or $script:targetRequests -ne 0) { throw 'Missing identity was not rejected before I/O' }
|
||||
$checks++
|
||||
$DiscoveryId='00000000-0000-4000-8000-000000000001';$ObservationVersion='version';$CommandId='command'
|
||||
Invoke-Expression $prefix
|
||||
if ($IpAddress -ne '198.51.100.10' -or $script:targetRequests -ne 1) { throw 'Unvalidated target retained' }
|
||||
$checks++
|
||||
Write-Output "Discovery PowerShell checks passed: $checks; mocked Hub only; zero miner sockets."
|
||||
@@ -0,0 +1,46 @@
|
||||
param(
|
||||
[Parameter(Mandatory=$true)][string]$BaselineRoot,
|
||||
[Parameter(Mandatory=$true)][string]$CachedPayloadPath,
|
||||
[Parameter(Mandatory=$true)][string]$MinerIp,
|
||||
[switch]$ExpectFenced
|
||||
)
|
||||
$ErrorActionPreference = "Stop"
|
||||
if ($env:MINEROPS_STAGE_A_SIMULATION -ne "1") {
|
||||
throw "Requires an explicitly isolated Stage A simulation; never use physical targets."
|
||||
}
|
||||
|
||||
# Supply a git-archived 2d89e80 Windows source directory, NOT a running installation.
|
||||
# Only AST function definitions execute; no startup, config loading, or task loops.
|
||||
$tokens = $null; $errors = $null
|
||||
$ast = [System.Management.Automation.Language.Parser]::ParseFile(
|
||||
(Join-Path $BaselineRoot "MinerOpsAgentService.ps1"), [ref]$tokens, [ref]$errors)
|
||||
if ($errors.Count) { throw "Baseline parser error" }
|
||||
foreach ($definition in $ast.FindAll({
|
||||
param($node)
|
||||
$node -is [System.Management.Automation.Language.FunctionDefinitionAst] -and
|
||||
$node.Parent -is [System.Management.Automation.Language.NamedBlockAst]
|
||||
}, $true)) {
|
||||
Invoke-Expression $definition.Extent.Text
|
||||
}
|
||||
. (Join-Path $BaselineRoot "MinerOpsControl.ps1")
|
||||
function Read-AgentConfig {
|
||||
return [pscustomobject]@{
|
||||
api3_account = "simulation"; api3_password = "simulation-only"; api3_port = 4433
|
||||
}
|
||||
}
|
||||
function Get-MinerOpsApi3Salt { return @{ salt = "simulation-only"; salt_response = @{} } }
|
||||
|
||||
$payload = Get-Content -LiteralPath $CachedPayloadPath -Raw | ConvertFrom-Json
|
||||
if ($payload.action -ne "suspend") { throw "This simulation expects the audited cached suspend command" }
|
||||
# Remap ONLY the target to the address of the dedicated disposable miner container.
|
||||
$payload.ip_address = $MinerIp
|
||||
$result = Run-MinerControlCommand -Payload $payload
|
||||
if ($ExpectFenced) {
|
||||
if ($result.command_accepted -ne $false -or $result.failure_code -ne "write_rejected") {
|
||||
throw "Old executor unexpectedly reached/accepted a simulated miner through the fence"
|
||||
}
|
||||
Write-Host "Fenced old cached command failed; independently verify simulator write count stayed zero."
|
||||
} else {
|
||||
if ($result.command_accepted -ne $true) { throw "Unfenced control simulation did not reproduce old execution" }
|
||||
Write-Host "Unfenced old cached command reproduced; simulated miner received service stop."
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Stage A containment: no current payload authorizes a physical write.
|
||||
|
||||
There is deliberately no environment-variable override. Re-enabling physical
|
||||
control requires a separately reviewed identity-binding protocol.
|
||||
"""
|
||||
|
||||
PHYSICAL_CONTROL_DISABLED = "physical_control_disabled_stage_a"
|
||||
UNSUPPORTED_COMMAND = "unsupported_command_schema_stage_a"
|
||||
READ_ONLY_COMMAND_TYPES = (
|
||||
"poll_known_miners_now",
|
||||
"discover_new_miners",
|
||||
"enrich_match_discovery",
|
||||
)
|
||||
|
||||
|
||||
def rejection_reason(command_type: str) -> str | None:
|
||||
if command_type == "miner_control":
|
||||
return PHYSICAL_CONTROL_DISABLED
|
||||
if command_type not in READ_ONLY_COMMAND_TYPES:
|
||||
return UNSUPPORTED_COMMAND
|
||||
return None
|
||||
|
||||
|
||||
def rejection_execution_outcome(row: dict) -> str:
|
||||
"""A current pending state alone says nothing about earlier delivery.
|
||||
|
||||
The repository supplies an explicit pre-transition evidence check while
|
||||
holding the queue row lock. Missing/ambiguous checks and result submissions
|
||||
are unknown. This describes queue delivery, never proof of physical outcome.
|
||||
"""
|
||||
if (row.get("previous_status") == "pending"
|
||||
and row.get("no_delivery_evidence") is True
|
||||
and row.get("delivery_reported", False) is False):
|
||||
return "not_dispatched"
|
||||
return "unknown"
|
||||
@@ -8,7 +8,10 @@ from psycopg.rows import dict_row
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
from app.agent_commands.exceptions import AgentCommandConflict, AgentCommandNotFound
|
||||
from app.agent_commands.policy import READ_ONLY_COMMAND_TYPES, rejection_execution_outcome, rejection_reason
|
||||
from app.config import settings
|
||||
from app.discovery.repository import DiscoveryRepository
|
||||
from fastapi import HTTPException
|
||||
|
||||
|
||||
ConnectionFactory = Callable[[], psycopg.Connection]
|
||||
@@ -19,6 +22,7 @@ WITH candidates AS (
|
||||
FROM minerops_agent_command_queue
|
||||
WHERE agent_id = %s
|
||||
AND command_status = 'pending'
|
||||
AND command_type = ANY(%s)
|
||||
ORDER BY created_at, id
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT %s
|
||||
@@ -32,6 +36,70 @@ RETURNING queue.id AS command_id, queue.command_type,
|
||||
queue.request_payload, queue.created_at
|
||||
"""
|
||||
|
||||
# Inspect existing durable history before changing status/completed_at/error.
|
||||
# A linked non-creation audit is conservatively ambiguous, even if timestamps
|
||||
# were lost. Legacy results need not agree on Agent ID to prevent a false claim
|
||||
# of non-delivery. Missing tables/permissions fail the transaction closed.
|
||||
NO_DELIVERY_EVIDENCE_SQL = """
|
||||
q.created_at IS NOT NULL
|
||||
AND q.claimed_at IS NULL AND q.completed_at IS NULL
|
||||
AND q.result_payload IS NULL AND COALESCE(q.error_message, '') = ''
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM minerops_agent_command_results r
|
||||
WHERE r.command_id = q.id
|
||||
OR (r.agent_id = q.agent_id AND NULLIF(btrim(r.command_id), '') IS NULL)
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM audit_logs a
|
||||
WHERE a.action <> 'miner.control.queued'
|
||||
AND (a.old_value_json ->> 'command_id' = q.id
|
||||
OR a.new_value_json ->> 'command_id' = q.id
|
||||
OR a.old_value_json ->> 'operation_id' = q.id
|
||||
OR a.new_value_json ->> 'operation_id' = q.id
|
||||
OR (a.entity_type = 'agent_command' AND a.entity_id::text = q.id))
|
||||
)
|
||||
"""
|
||||
|
||||
RETIRE_COMMANDS_SQL = f"""
|
||||
WITH candidates AS (
|
||||
SELECT q.id, q.command_status, ({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
|
||||
FROM minerops_agent_command_queue q
|
||||
WHERE agent_id = %s AND command_status IN ('pending', 'claimed')
|
||||
AND NOT (command_type = ANY(%s))
|
||||
ORDER BY created_at, id
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1000
|
||||
)
|
||||
UPDATE minerops_agent_command_queue AS queue
|
||||
SET command_status = 'cancelled', completed_at = COALESCE(queue.completed_at, now()),
|
||||
error_message = concat_ws(chr(10), NULLIF(queue.error_message, ''), CASE WHEN queue.command_type = 'miner_control'
|
||||
THEN 'physical_control_disabled_stage_a'
|
||||
ELSE 'unsupported_command_schema_stage_a' END)
|
||||
FROM candidates WHERE queue.id = candidates.id
|
||||
RETURNING queue.id, queue.agent_id, queue.command_type,
|
||||
candidates.command_status AS previous_status, candidates.no_delivery_evidence, queue.error_message
|
||||
"""
|
||||
|
||||
RETIRE_SCHEDULE_COMMANDS_SQL = f"""
|
||||
WITH candidates AS (
|
||||
SELECT q.id, q.command_status, ({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
|
||||
FROM minerops_agent_command_queue q
|
||||
WHERE agent_id = %s AND request_payload ->> 'scheduled_action_id' = %s
|
||||
AND command_status IN ('pending', 'claimed')
|
||||
AND NOT (command_type = ANY(%s))
|
||||
ORDER BY id
|
||||
FOR UPDATE
|
||||
)
|
||||
UPDATE minerops_agent_command_queue AS queue
|
||||
SET command_status = 'cancelled', completed_at = COALESCE(queue.completed_at, now()),
|
||||
error_message = concat_ws(chr(10), NULLIF(queue.error_message, ''), CASE WHEN queue.command_type = 'miner_control'
|
||||
THEN 'physical_control_disabled_stage_a'
|
||||
ELSE 'unsupported_command_schema_stage_a' END)
|
||||
FROM candidates WHERE queue.id = candidates.id
|
||||
RETURNING queue.id, queue.agent_id, queue.command_type,
|
||||
candidates.command_status AS previous_status, candidates.no_delivery_evidence
|
||||
"""
|
||||
|
||||
|
||||
def default_connection_factory() -> psycopg.Connection:
|
||||
return psycopg.connect(
|
||||
@@ -54,8 +122,29 @@ class AgentCommandRepository:
|
||||
connection = self._connection_factory()
|
||||
try:
|
||||
with connection.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute(CLAIM_COMMANDS_SQL, (agent_id, limit))
|
||||
DiscoveryRepository.lock(cur)
|
||||
self._retire_unsupported(cur, agent_id)
|
||||
discovery = DiscoveryRepository(self._connection_factory)
|
||||
cur.execute(f"""SELECT q.*, ({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
|
||||
FROM minerops_agent_command_queue q WHERE agent_id=%s
|
||||
AND command_status='pending' AND command_type=ANY(%s)
|
||||
ORDER BY created_at,id FOR UPDATE SKIP LOCKED LIMIT 1000""",
|
||||
(agent_id,list(READ_ONLY_COMMAND_TYPES)))
|
||||
pending=list(cur.fetchall())
|
||||
for command in pending:
|
||||
try:
|
||||
discovery.validate_command(cur,agent_id,command)
|
||||
except HTTPException as exc:
|
||||
cur.execute("""UPDATE minerops_agent_command_queue SET command_status='cancelled',
|
||||
completed_at=COALESCE(completed_at,now()), error_message=concat_ws(chr(10),NULLIF(error_message,''),%s::text)
|
||||
WHERE id=%s""",(exc.detail,command['id']))
|
||||
DiscoveryRepository.audit(cur,'discovery.command_rejected',None,
|
||||
{'command_id':command['id'],'agent_id':agent_id,'reason':exc.detail,
|
||||
'execution_outcome':rejection_execution_outcome({**command,'previous_status':'pending'})})
|
||||
cur.execute(CLAIM_COMMANDS_SQL, (agent_id, list(READ_ONLY_COMMAND_TYPES), limit))
|
||||
rows = [dict(row) for row in cur.fetchall()]
|
||||
for row in rows:
|
||||
discovery.validate_command(cur,agent_id,row)
|
||||
connection.commit()
|
||||
return rows
|
||||
except Exception:
|
||||
@@ -64,6 +153,105 @@ class AgentCommandRepository:
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
@staticmethod
|
||||
def _audit_rejection(
|
||||
cur, row: dict, *, actor_user_id: str | None = None,
|
||||
source: str = "agent_poll",
|
||||
) -> None:
|
||||
# Queue IDs can be legacy text, not UUIDs. Preserve them in JSON rather
|
||||
# than coercing them into audit_logs.entity_id. Never log raw payloads.
|
||||
cur.execute(
|
||||
"""INSERT INTO audit_logs
|
||||
(actor_user_id, action, entity_type, old_value_json, new_value_json)
|
||||
VALUES (%s, 'agent.command.rejected', 'agent_command', %s, %s)""",
|
||||
(actor_user_id, Jsonb({"command_id": row["id"], "agent_id": row["agent_id"],
|
||||
"status": row["previous_status"]}),
|
||||
Jsonb({"status": "cancelled", "reason": rejection_reason(row["command_type"]),
|
||||
"source": source,
|
||||
"execution_outcome": rejection_execution_outcome(row)})),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _retire_unsupported(
|
||||
cls, cur, agent_id: str, *, source: str = "agent_poll",
|
||||
actor_user_id: str | None = None,
|
||||
) -> int:
|
||||
cur.execute(RETIRE_COMMANDS_SQL, (agent_id, list(READ_ONLY_COMMAND_TYPES)))
|
||||
rows = list(cur.fetchall())
|
||||
for row in rows:
|
||||
cls._audit_rejection(cur, row, source=source, actor_user_id=actor_user_id)
|
||||
return len(rows)
|
||||
|
||||
def retire_unsupported(self, agent_id: str, *, actor_user_id: str) -> int:
|
||||
"""Audited cutover batch, including offline Agents; never claim any work.
|
||||
|
||||
A zero batch can mean locked rows were skipped. Operators must separately
|
||||
verify no unsupported live rows remain before declaring a drain complete.
|
||||
"""
|
||||
if not actor_user_id:
|
||||
raise ValueError("Cutover retirement requires an accountable operator actor")
|
||||
connection = self._connection_factory()
|
||||
try:
|
||||
with connection.cursor(row_factory=dict_row) as cur:
|
||||
count = self._retire_unsupported(cur, agent_id, source="stage_a_cutover",
|
||||
actor_user_id=actor_user_id)
|
||||
connection.commit()
|
||||
return count
|
||||
except Exception:
|
||||
connection.rollback()
|
||||
raise
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def retire_schedule(self, schedule_id: str, *, actor_user_id: str, operation: str) -> int:
|
||||
"""Cancel live unsupported work and retain schedule history atomically.
|
||||
|
||||
The stored schedule determines Agent scope. Delete retires the schedule;
|
||||
neither operation erases evidence or changes terminal command history.
|
||||
"""
|
||||
if not actor_user_id or operation not in {"cancel", "delete"}:
|
||||
raise ValueError("An authenticated actor and cancel/delete operation are required")
|
||||
connection = self._connection_factory()
|
||||
try:
|
||||
with connection.cursor(row_factory=dict_row) as cur:
|
||||
# Serialize retries and cancel/delete workers for this schedule.
|
||||
cur.execute(
|
||||
"""SELECT id, agent_id, status FROM minerops_scheduled_bulk_actions
|
||||
WHERE id = %s FOR UPDATE""", (schedule_id,),
|
||||
)
|
||||
schedule = cur.fetchone()
|
||||
if not schedule:
|
||||
raise AgentCommandNotFound("Scheduled action not found")
|
||||
cur.execute(RETIRE_SCHEDULE_COMMANDS_SQL,
|
||||
(schedule["agent_id"], schedule_id, list(READ_ONLY_COMMAND_TYPES)))
|
||||
rows = list(cur.fetchall())
|
||||
for row in rows:
|
||||
self._audit_rejection(cur, row, actor_user_id=actor_user_id,
|
||||
source=f"schedule_{operation}")
|
||||
if schedule["status"] in {"pending", "queued", "running"}:
|
||||
cur.execute(
|
||||
"""UPDATE minerops_scheduled_bulk_actions
|
||||
SET status = 'cancelled', updated_at = now(),
|
||||
last_error = concat_ws(chr(10), NULLIF(last_error, ''),
|
||||
'physical_control_disabled_stage_a')
|
||||
WHERE id = %s""", (schedule_id,),
|
||||
)
|
||||
cur.execute(
|
||||
"""INSERT INTO audit_logs
|
||||
(actor_user_id, action, entity_type, old_value_json, new_value_json)
|
||||
VALUES (%s, 'scheduled_action.retired', 'scheduled_action', %s, %s)""",
|
||||
(actor_user_id, Jsonb(dict(schedule)),
|
||||
Jsonb({"status": "cancelled", "operation": operation,
|
||||
"reason": "physical_control_disabled_stage_a"})),
|
||||
)
|
||||
connection.commit()
|
||||
return len(rows)
|
||||
except Exception:
|
||||
connection.rollback()
|
||||
raise
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def get_command(self, command_id: str) -> dict | None:
|
||||
connection = self._connection_factory()
|
||||
try:
|
||||
@@ -91,6 +279,32 @@ class AgentCommandRepository:
|
||||
connection = self._connection_factory()
|
||||
try:
|
||||
with connection.cursor(row_factory=dict_row) as cur:
|
||||
# Enforce the same boundary on results: an old claimed entry
|
||||
# cannot be promoted by an old Agent, retry, or direct caller.
|
||||
cur.execute(
|
||||
f"""SELECT q.id, q.agent_id, q.command_type, q.command_status AS previous_status,
|
||||
({NO_DELIVERY_EVIDENCE_SQL}) AS no_delivery_evidence
|
||||
FROM minerops_agent_command_queue q
|
||||
WHERE q.id = %s AND q.agent_id = %s FOR UPDATE""",
|
||||
(command_id, agent_id),
|
||||
)
|
||||
existing = cur.fetchone()
|
||||
if existing and rejection_reason(existing["command_type"]):
|
||||
reason = rejection_reason(existing["command_type"])
|
||||
if existing["previous_status"] in {"pending", "claimed"}:
|
||||
cur.execute(
|
||||
"""UPDATE minerops_agent_command_queue
|
||||
SET command_status = 'cancelled',
|
||||
error_message = concat_ws(chr(10), NULLIF(error_message, ''), %s::text),
|
||||
completed_at = COALESCE(completed_at, now()) WHERE id = %s""",
|
||||
(reason, command_id),
|
||||
)
|
||||
# A result submission itself reports possible delivery, even
|
||||
# when earlier durable claim metadata is absent.
|
||||
existing["delivery_reported"] = True
|
||||
self._audit_rejection(cur, existing, source="agent_result")
|
||||
connection.commit()
|
||||
raise AgentCommandConflict(reason)
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE minerops_agent_command_queue
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
"""HTTP boundary for authenticated Agent command transport."""
|
||||
|
||||
import logging
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
@@ -18,8 +16,6 @@ from app.agent_commands.models import (
|
||||
)
|
||||
from app.agent_commands.service import AgentCommandService
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/agent/commands", tags=["agent-commands"])
|
||||
AgentIdentity = Annotated[str, Depends(require_agent_identity)]
|
||||
AgentService = Annotated[AgentCommandService, Depends(get_agent_command_service)]
|
||||
@@ -29,14 +25,7 @@ AgentService = Annotated[AgentCommandService, Depends(get_agent_command_service)
|
||||
def claim_pending_commands(
|
||||
request: Request, agent_id: AgentIdentity, service: AgentService
|
||||
) -> AgentCommandListResponse:
|
||||
before_claim = getattr(
|
||||
request.app.state, "agent_command_before_claim", None
|
||||
)
|
||||
if callable(before_claim):
|
||||
try:
|
||||
before_claim()
|
||||
except Exception:
|
||||
logger.exception("Scheduled command materialization failed before Agent claim")
|
||||
del request
|
||||
return AgentCommandListResponse(commands=service.claim(agent_id))
|
||||
|
||||
|
||||
|
||||
@@ -2,19 +2,16 @@
|
||||
|
||||
import re
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.agent_commands.exceptions import (
|
||||
AgentCommandConflict,
|
||||
AgentCommandNotFound,
|
||||
AgentResultValidationError,
|
||||
)
|
||||
from app.agent_commands.models import (
|
||||
AgentCommand,
|
||||
AgentCommandResultRequest,
|
||||
MinerControlResultPayload,
|
||||
)
|
||||
from app.agent_commands.repository import AgentCommandRepository
|
||||
from app.agent_commands.policy import rejection_reason
|
||||
|
||||
|
||||
class AgentCommandService:
|
||||
@@ -25,6 +22,7 @@ class AgentCommandService:
|
||||
return [
|
||||
AgentCommand.model_validate(row)
|
||||
for row in self.repository.claim_pending(agent_id)
|
||||
if rejection_reason(row["command_type"]) is None
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
@@ -42,6 +40,14 @@ class AgentCommandService:
|
||||
raise AgentCommandNotFound("Agent command not found")
|
||||
if command["agent_id"] != agent_id:
|
||||
raise AgentCommandConflict("Agent command belongs to another Agent")
|
||||
reason = rejection_reason(command["command_type"])
|
||||
if reason:
|
||||
# Repository repeats this check under a row lock and records the
|
||||
# cancellation atomically; never accept an old physical result.
|
||||
self.repository.finish_command(
|
||||
request.command_id, agent_id, "cancelled", {}, reason
|
||||
)
|
||||
raise AgentCommandConflict(reason)
|
||||
if command["command_status"] != "claimed":
|
||||
raise AgentCommandConflict(
|
||||
f"Agent command is {command['command_status']}, not claimed"
|
||||
@@ -49,34 +55,6 @@ class AgentCommandService:
|
||||
|
||||
result_payload = request.result_payload
|
||||
error_message = self._legacy_error(request.error_message)
|
||||
if command["command_type"] == "miner_control":
|
||||
try:
|
||||
result = MinerControlResultPayload.model_validate(request.result_payload)
|
||||
except ValidationError as exc:
|
||||
raise AgentResultValidationError(
|
||||
"Invalid typed miner-control result"
|
||||
) from exc
|
||||
if request.status == "completed" and not result.command_accepted:
|
||||
raise AgentResultValidationError(
|
||||
"Completed miner-control results must be accepted"
|
||||
)
|
||||
if request.status == "failed" and result.command_accepted:
|
||||
raise AgentResultValidationError(
|
||||
"Failed miner-control results cannot be accepted"
|
||||
)
|
||||
action = (command.get("request_payload") or {}).get("action")
|
||||
if (
|
||||
action in {"suspend", "resume"}
|
||||
and result.command_accepted
|
||||
and result.verification_status != "unconfirmed"
|
||||
):
|
||||
raise AgentResultValidationError(
|
||||
"Accepted suspend and resume results must remain unconfirmed"
|
||||
)
|
||||
result_payload = result.model_dump(mode="json")
|
||||
error_message = (
|
||||
result.failure_code.replace("_", " ") if result.failure_code else None
|
||||
)
|
||||
|
||||
self.repository.finish_command(
|
||||
request.command_id,
|
||||
|
||||
@@ -108,6 +108,11 @@ class AgentZoneRoutingContext(BaseModel):
|
||||
routing_reason: str | None = None
|
||||
|
||||
|
||||
class AgentDiscoveryMinerMatch(AgentZoneRoutingContext):
|
||||
miner_id: UUID | None = None
|
||||
matched_by: str | None = None
|
||||
|
||||
|
||||
class MinerZoneAssociationResult(BaseModel):
|
||||
agent_zone_id: UUID | None = None
|
||||
routing_reason: str | None = None
|
||||
|
||||
@@ -204,6 +204,8 @@ class AgentZoneRepository:
|
||||
ip_address = EXCLUDED.ip_address,
|
||||
last_checkin_at = now(),
|
||||
last_payload = EXCLUDED.last_payload
|
||||
WHERE minerops_site_agents.site_id IS NOT DISTINCT FROM EXCLUDED.site_id
|
||||
AND minerops_site_agents.agent_zone_id IS NOT DISTINCT FROM EXCLUDED.agent_zone_id
|
||||
""",
|
||||
(
|
||||
agent_id,
|
||||
@@ -216,6 +218,8 @@ class AgentZoneRepository:
|
||||
Jsonb(payload),
|
||||
),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
raise AgentZoneConflict("Agent assignment changed concurrently")
|
||||
connection.commit()
|
||||
except UniqueViolation as exc:
|
||||
connection.rollback()
|
||||
@@ -294,6 +298,20 @@ class AgentZoneRepository:
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def find_miner_in_zone(self, *, agent_id: str, agent_zone_id: UUID,
|
||||
mac_address: str | None, ip_address: str | None) -> dict | None:
|
||||
"""Retained helper uses the same atomic authority/collision boundary as HTTP."""
|
||||
from app.discovery.repository import DiscoveryRepository, reject
|
||||
repository = DiscoveryRepository(self._connection_factory)
|
||||
with repository.transaction() as cur:
|
||||
context = repository.context(cur, agent_id)
|
||||
if context['agent_zone_id'] != agent_zone_id:
|
||||
reject('discovery_stale_authority')
|
||||
if not mac_address and not ip_address:
|
||||
return None
|
||||
row = repository.match(cur, context, {'mac_address': mac_address, 'ip_address': ip_address})
|
||||
return {'id': row['id'], 'matched_by': 'mac' if mac_address else 'ip'} if row else None
|
||||
|
||||
def assign_miners_to_zone(
|
||||
self, miner_ids: Iterable[UUID], agent_zone_id: UUID
|
||||
) -> dict[UUID, str]:
|
||||
@@ -326,15 +344,7 @@ class AgentZoneRepository:
|
||||
elif miner["site_id"] not in {None, zone["site_id"]}:
|
||||
outcomes[miner_id] = "site_conflict"
|
||||
else:
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE miners
|
||||
SET site_id = %s, agent_zone_id = %s, updated_at = now()
|
||||
WHERE id = %s AND agent_zone_id IS NULL
|
||||
""",
|
||||
(zone["site_id"], agent_zone_id, miner_id),
|
||||
)
|
||||
outcomes[miner_id] = "assigned"
|
||||
outcomes[miner_id] = "miner_zone_unassigned"
|
||||
connection.commit()
|
||||
return outcomes
|
||||
except Exception:
|
||||
|
||||
@@ -10,6 +10,7 @@ from app.agent_zones.exceptions import (
|
||||
from app.agent_zones.models import (
|
||||
AgentCheckinRequest,
|
||||
AgentCheckinResponse,
|
||||
AgentDiscoveryMinerMatch,
|
||||
AgentZoneRoutingContext,
|
||||
AgentZoneCreateRequest,
|
||||
AgentZoneSummary,
|
||||
@@ -81,6 +82,8 @@ class AgentZoneService:
|
||||
self.validate_agent_identity(authenticated_agent_id, request.agent_id)
|
||||
|
||||
existing = self.repository.get_agent(authenticated_agent_id)
|
||||
if not existing or existing.get("agent_zone_id") is None:
|
||||
raise AgentZoneValidationError("Agent requires an administrator-established Zone assignment")
|
||||
requested_site = request.site_id
|
||||
requested_zone = request.agent_zone_id
|
||||
final_site = requested_site or (existing or {}).get("site_id")
|
||||
@@ -199,6 +202,30 @@ class AgentZoneService:
|
||||
],
|
||||
}
|
||||
|
||||
def match_discovered_miner(
|
||||
self,
|
||||
agent_id: str,
|
||||
*,
|
||||
mac_address: str | None,
|
||||
ip_address: str | None,
|
||||
recorded_zone_id: UUID | None = None,
|
||||
) -> AgentDiscoveryMinerMatch:
|
||||
routing = self.routing_context(agent_id, recorded_zone_id)
|
||||
if routing.routing_reason or routing.agent_zone_id is None:
|
||||
return AgentDiscoveryMinerMatch(**routing.model_dump())
|
||||
|
||||
match = self.repository.find_miner_in_zone(
|
||||
agent_id=agent_id,
|
||||
agent_zone_id=routing.agent_zone_id,
|
||||
mac_address=mac_address,
|
||||
ip_address=ip_address,
|
||||
)
|
||||
return AgentDiscoveryMinerMatch(
|
||||
**routing.model_dump(),
|
||||
miner_id=match["id"] if match else None,
|
||||
matched_by=match["matched_by"] if match else None,
|
||||
)
|
||||
|
||||
def associate_miners_from_agent(
|
||||
self, agent_id: str, miner_ids: list[UUID]
|
||||
) -> MinerZoneAssociationResult:
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Agent discovery isolation; physical miner writes are never authorized here."""
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Exact administrative CSV selection, inside the caller's locked transaction.
|
||||
|
||||
CSV metadata never establishes ownership or replaces board identity. Every supplied
|
||||
identity is a constraint; a failed strong identifier never falls back to an IP.
|
||||
"""
|
||||
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.discovery.repository import address, mac
|
||||
|
||||
|
||||
def resolve_inventory_row(cur, row, row_number, selected_zone=None):
|
||||
def fail(message, status=409):
|
||||
raise HTTPException(status_code=status, detail=f"CSV row {row_number}: {message}")
|
||||
|
||||
def field(names, normalize=lambda v: v):
|
||||
values = []
|
||||
for name in names:
|
||||
value = row.get(name)
|
||||
if value is not None and str(value).strip():
|
||||
try:
|
||||
values.append(normalize(str(value).strip()))
|
||||
except (ValueError, HTTPException):
|
||||
fail(f"invalid {names[0]}", 422)
|
||||
if len(set(values)) > 1:
|
||||
fail(f"conflicting {names[0]} columns")
|
||||
return values[0] if values else None
|
||||
|
||||
def uuid(value):
|
||||
return str(UUID(value))
|
||||
|
||||
mid = field(["Miner ID", "Miner UUID", "miner_id", "id"], uuid)
|
||||
zone = field(["Agent Zone ID", "Zone ID", "agent_zone_id", "zone_id"], uuid)
|
||||
site = field(["Site ID", "site_id"], uuid)
|
||||
ip = field(["IP", "IP Address", "ip", "ip_address"], address)
|
||||
current_mac = field(["MAC Addr", "MAC", "Mac", "mac", "mac_address"], mac)
|
||||
serial = field(["Serial", "Serial Number", "SN", "Miner SN", "miner_sn", "serial_number"])
|
||||
if selected_zone:
|
||||
try:
|
||||
selected_zone = uuid(selected_zone)
|
||||
except ValueError:
|
||||
fail("invalid selected Agent Zone ID", 422)
|
||||
if zone and zone != selected_zone:
|
||||
fail("row Zone disagrees with selected Zone")
|
||||
zone = selected_zone
|
||||
if zone:
|
||||
cur.execute("SELECT site_id::text FROM agent_zones WHERE id=%s", (zone,))
|
||||
zone_row = cur.fetchone()
|
||||
if not zone_row:
|
||||
fail("selected Zone does not exist", 422)
|
||||
if site and site != zone_row["site_id"]:
|
||||
fail("Site and Zone disagree")
|
||||
if not (mid or current_mac or serial or (ip and zone)):
|
||||
fail("provide Miner ID, current MAC, serial, or IP plus Agent Zone ID", 422)
|
||||
|
||||
selectors = []
|
||||
if mid:
|
||||
selectors.append(("id=%s::uuid", mid))
|
||||
if current_mac:
|
||||
selectors.append(("regexp_replace(lower(current_mac), '[^0-9a-f]', '', 'g')=%s", current_mac.replace(":", "").lower()))
|
||||
if serial:
|
||||
selectors.append(("miner_sn=%s", serial))
|
||||
# IP is only a namespace-scoped selector when no stronger identity is supplied.
|
||||
if not selectors:
|
||||
selectors.append(("current_ip=%s::inet AND agent_zone_id=%s::uuid", (ip, zone)))
|
||||
|
||||
candidates = {}
|
||||
for predicate, value in selectors:
|
||||
params = value if isinstance(value, tuple) else (value,)
|
||||
cur.execute(
|
||||
"SELECT id::text, site_id::text, agent_zone_id::text, "
|
||||
"host(current_ip) AS current_ip, current_mac, miner_sn, model, miner_type, firmware_version "
|
||||
"FROM miners WHERE " + predicate + " ORDER BY id LIMIT 2 FOR UPDATE",
|
||||
params,
|
||||
)
|
||||
matches = cur.fetchall()
|
||||
if len(matches) > 1:
|
||||
fail("ambiguous inventory identity; supply exact Miner ID and consistent identifiers")
|
||||
candidates.update({m["id"]: m for m in matches})
|
||||
if not candidates:
|
||||
return None
|
||||
if len(candidates) != 1:
|
||||
fail("identifiers select different inventory records")
|
||||
miner = next(iter(candidates.values()))
|
||||
if (
|
||||
(mid and mid != miner["id"])
|
||||
or (zone and zone != miner["agent_zone_id"])
|
||||
or (site and site != miner["site_id"])
|
||||
or (ip and ip != miner["current_ip"])
|
||||
or (serial and serial != miner["miner_sn"])
|
||||
or (current_mac and current_mac != mac(miner["current_mac"]))
|
||||
):
|
||||
fail("identifiers disagree with current inventory; refresh selection before importing")
|
||||
return miner
|
||||
@@ -0,0 +1,42 @@
|
||||
"""Bounded HTTP adapter; authentication remains on the registered routes."""
|
||||
|
||||
import json
|
||||
from fastapi import HTTPException
|
||||
from starlette.concurrency import run_in_threadpool
|
||||
import psycopg
|
||||
from app.discovery.repository import DiscoveryRepository
|
||||
|
||||
MAX_BODY = 1024 * 1024
|
||||
|
||||
|
||||
async def dispatch(method, request=None, **kwargs):
|
||||
if request is not None:
|
||||
chunks = []
|
||||
size = 0
|
||||
async for chunk in request.stream():
|
||||
size += len(chunk)
|
||||
if size > MAX_BODY:
|
||||
raise HTTPException(413, "discovery_payload_too_large")
|
||||
chunks.append(chunk)
|
||||
try:
|
||||
|
||||
def invalid_constant(_):
|
||||
raise ValueError("Non-finite JSON number")
|
||||
|
||||
kwargs["body"] = json.loads(
|
||||
b"".join(chunks), parse_constant=invalid_constant
|
||||
)
|
||||
except (ValueError, UnicodeDecodeError, RecursionError):
|
||||
raise HTTPException(422, "discovery_invalid_json")
|
||||
try:
|
||||
return await run_in_threadpool(getattr(DiscoveryRepository(), method), **kwargs)
|
||||
except (
|
||||
psycopg.errors.LockNotAvailable,
|
||||
psycopg.errors.QueryCanceled,
|
||||
psycopg.errors.DeadlockDetected,
|
||||
):
|
||||
raise HTTPException(409, "discovery_concurrent_change_retry")
|
||||
except psycopg.DataError:
|
||||
raise HTTPException(422, "discovery_invalid_value")
|
||||
except psycopg.IntegrityError:
|
||||
raise HTTPException(409, "discovery_identity_conflict")
|
||||
File diff suppressed because it is too large.
Load diff
@@ -708,7 +708,10 @@ def machine_agent_request(path: str) -> bool:
|
||||
"/api/agent/pool-snapshots",
|
||||
"/api/agent/commands/pending",
|
||||
"/api/agent/commands/result",
|
||||
"/api/agent/commands/pending-disabled-old",
|
||||
"/api/agent/commands/result-disabled-old",
|
||||
"/api/agent/discoveries",
|
||||
"/api/agent/discovery-target",
|
||||
"/api/agent/miner-enrichment",
|
||||
"/api/agent/miner-enrichment-v2",
|
||||
"/api/agent/miner-enrichment-v3",
|
||||
@@ -760,6 +763,19 @@ def component_request(path: str) -> bool:
|
||||
|
||||
def required_api_permission(method: str, path: str) -> str | None:
|
||||
parts = path.split("/")
|
||||
if method == "POST" and path == "/api/dashboard/m30-command-test-lab/run":
|
||||
return "miners.control"
|
||||
if method == "POST" and path == "/api/tools/whatsminer-read-test":
|
||||
return "miners.control"
|
||||
if method == "POST" and path in {
|
||||
"/api/dashboard/scheduled-bulk-actions",
|
||||
"/api/dashboard/scheduled-bulk-actions/process-due",
|
||||
}:
|
||||
return "miners.control"
|
||||
if (method == "POST" and len(parts) == 6
|
||||
and parts[:4] == ["", "api", "dashboard", "scheduled-bulk-actions"]
|
||||
and parts[5] in {"cancel", "delete"}):
|
||||
return "miners.control"
|
||||
if (
|
||||
len(parts) >= 5
|
||||
and parts[:3] == ["", "api", "miners"]
|
||||
@@ -811,6 +827,8 @@ def required_api_permission(method: str, path: str) -> str | None:
|
||||
return "repairs.complete"
|
||||
if method == "GET" and repair_timeline_request(path):
|
||||
return "repairs.read"
|
||||
if method == "GET" and path in {"/api/agent/discoveries/summary", "/api/agent/status-summary"}:
|
||||
return "agents.view"
|
||||
if method == "GET":
|
||||
if path.startswith("/api/system/"):
|
||||
return "system_health.view"
|
||||
|
||||
+346
-5531
File diff suppressed because it is too large.
Load diff
@@ -1,7 +1,6 @@
|
||||
"""Business rules for typed, Agent-routed miner controls."""
|
||||
|
||||
from typing import Any
|
||||
from uuid import UUID, uuid4
|
||||
from uuid import UUID
|
||||
|
||||
from app.miner_control.exceptions import (
|
||||
ControlledMinerNotFound,
|
||||
@@ -16,6 +15,7 @@ from app.miner_control.models import (
|
||||
MinerControlProjection,
|
||||
)
|
||||
from app.miner_control.repository import MinerControlRepository
|
||||
from app.agent_commands.policy import PHYSICAL_CONTROL_DISABLED
|
||||
|
||||
|
||||
KNOWN_MODES = {"low", "normal", "high"}
|
||||
@@ -29,33 +29,8 @@ class MinerControlService:
|
||||
|
||||
@staticmethod
|
||||
def _capability(context: dict) -> ControlCapability:
|
||||
if not context.get("agent_zone_id"):
|
||||
return ControlCapability(state="unavailable", reason="agent_zone_unassigned")
|
||||
if context.get("agent_zone_exists") is not True:
|
||||
return ControlCapability(state="unavailable", reason="agent_zone_missing")
|
||||
if context.get("agent_zone_enabled") is not True:
|
||||
return ControlCapability(state="unavailable", reason="agent_zone_disabled")
|
||||
if not context.get("agent_id"):
|
||||
return ControlCapability(state="unavailable", reason="agent_unavailable")
|
||||
if not context.get("current_ip"):
|
||||
return ControlCapability(state="unavailable", reason="current_ip_missing")
|
||||
if context.get("api3_supported") is False:
|
||||
return ControlCapability(state="unavailable", reason="api3_unsupported")
|
||||
if context.get("api3_supported") is not True:
|
||||
return ControlCapability(state="unknown", reason="api3_support_unknown")
|
||||
|
||||
payload = context.get("last_payload") or {}
|
||||
if payload.get("control_available") is False:
|
||||
reason = payload.get("control_unavailable_reason")
|
||||
return ControlCapability(
|
||||
state="unavailable",
|
||||
reason=reason if reason == "credentials_missing" else "agent_unavailable",
|
||||
)
|
||||
if payload.get("control_available") is not True:
|
||||
return ControlCapability(
|
||||
state="unknown", reason="agent_capability_unknown"
|
||||
)
|
||||
return ControlCapability(state="supported")
|
||||
# Credentials and Zone routing are not physical identity authorization.
|
||||
return ControlCapability(state="unavailable", reason=PHYSICAL_CONTROL_DISABLED)
|
||||
|
||||
@staticmethod
|
||||
def _operation(miner_id: UUID, row: dict) -> MinerControlOperation:
|
||||
@@ -120,50 +95,4 @@ class MinerControlService:
|
||||
) -> MinerControlAccepted:
|
||||
if not actor_user_id:
|
||||
raise MinerControlActorRequired("Authenticated actor UUID is required")
|
||||
context = self.repository.get_control_context(miner_id)
|
||||
if not context:
|
||||
raise ControlledMinerNotFound("Miner not found")
|
||||
capability = self._capability(context)
|
||||
if capability.state != "supported":
|
||||
raise MinerControlUnavailable(
|
||||
capability.reason or "control_capability_not_supported"
|
||||
)
|
||||
|
||||
runtime = str(context.get("runtime_state") or "").strip().lower()
|
||||
if action == "suspend" and runtime not in RUNNING_STATES:
|
||||
raise MinerControlUnavailable("miner_not_factually_running")
|
||||
if action == "resume" and runtime not in STOPPED_STATES:
|
||||
raise MinerControlUnavailable("miner_not_factually_stopped")
|
||||
if action == "performance_mode" and requested_value not in KNOWN_MODES:
|
||||
raise MinerControlUnavailable("invalid_performance_mode")
|
||||
|
||||
operation_id = str(uuid4())
|
||||
payload: dict[str, Any] = {
|
||||
"schema_version": 1,
|
||||
"miner_id": str(miner_id),
|
||||
"action": action,
|
||||
"requested_value": requested_value,
|
||||
"ip_address": context["current_ip"],
|
||||
}
|
||||
operation = self.repository.enqueue_with_audit(
|
||||
operation_id=operation_id,
|
||||
miner_id=miner_id,
|
||||
agent_id=context["agent_id"],
|
||||
actor_user_id=actor_user_id,
|
||||
actor_username=actor_username,
|
||||
request_payload=payload,
|
||||
audit_payload={
|
||||
"operation_id": operation_id,
|
||||
"miner_id": str(miner_id),
|
||||
"action": action,
|
||||
"requested_value": requested_value,
|
||||
},
|
||||
)
|
||||
return MinerControlAccepted(
|
||||
operation_id=operation["operation_id"],
|
||||
miner_id=miner_id,
|
||||
action=action,
|
||||
requested_value=requested_value,
|
||||
status=operation["status"],
|
||||
created_at=operation["created_at"],
|
||||
)
|
||||
raise MinerControlUnavailable(PHYSICAL_CONTROL_DISABLED)
|
||||
@@ -0,0 +1,49 @@
|
||||
// Actual Chromium; every HTTP request is fulfilled or aborted, with disposable data only.
|
||||
const {chromium}=require('playwright');
|
||||
const fs=require('fs');const assert=require('node:assert/strict');
|
||||
(async()=>{
|
||||
const browser=await chromium.launch({headless:true,args:['--no-sandbox','--disable-background-networking']});
|
||||
try {
|
||||
const context=await browser.newContext({serviceWorkers:'block'});
|
||||
const page=await context.newPage();
|
||||
const marker='<img src="data:," onerror="window.__discoveryAudit=1">';
|
||||
const commands=[];const unexpected=[];
|
||||
const observations=[0,1].map(n=>({id:'00000000-0000-4000-8000-00000000000'+(n+1),agent_id:'Agent '+n,agent_zone_id:'Zone '+n,version:'2026-01-01T00:00:00.00000'+n+'+00:00',created_at:'2026-01-01T00:00:00Z',ip_address:'198.51.100.10',room_name:marker,model:marker,mac_address:marker,discovery_status:'new'}));
|
||||
await page.route('**/*',async route=>{
|
||||
const url=new URL(route.request().url());
|
||||
if(url.origin!=='http://minerops.invalid'){unexpected.push(url.href);return route.abort();}
|
||||
const path=url.pathname;
|
||||
if(path==='/agent/control')return route.fulfill({contentType:'text/html',body:fs.readFileSync(process.argv[2],'utf8')});
|
||||
if(path==='/api/agent/status-summary')return route.fulfill({json:{live_database_counts:{total:2,online:2},latest_snapshot:{},agent:{agent_id:marker,hostname:marker,version:marker},recent_commands:[{command_type:marker,command_status:'failed',requested_by_username:marker,error_message:marker}]}});
|
||||
if(path==='/api/agent/discoveries')return route.fulfill({json:{discoveries:observations}});
|
||||
if(path==='/api/agent/discoveries/summary')return route.fulfill({json:{total:2,counts:{new:2}}});
|
||||
if(path==='/api/agent/discoveries/review'||path==='/api/agent/discoveries/enrich-match'){
|
||||
commands.push({path,body:route.request().postDataJSON()});return route.fulfill({json:{ok:true,message:'Saved'}});
|
||||
}
|
||||
return route.abort();
|
||||
});
|
||||
await context.addInitScript(()=>localStorage.setItem('minerops_token','disposable-browser-token'));
|
||||
page.on('dialog',dialog=>dialog.accept());
|
||||
await page.goto('http://minerops.invalid/agent/control');
|
||||
await page.waitForSelector('#discoveriesBody tr button');
|
||||
if(process.argv[3]==='baseline'){
|
||||
await page.waitForFunction(()=>window.__discoveryAudit===1);
|
||||
console.log('Baseline positive control: harmless stored event handler executed.');
|
||||
} else {
|
||||
assert.equal(await page.evaluate(()=>window.__discoveryAudit),undefined);
|
||||
assert.equal(await page.locator('#discoveriesBody img,#commandsBody img,[onerror]').count(),0);
|
||||
assert.ok((await page.locator('#discoveriesBody').innerText()).includes(marker));
|
||||
await page.locator('#discoveriesBody tr').nth(0).getByRole('button',{name:'Add as New',exact:true}).click();
|
||||
await page.waitForFunction(()=>document.getElementById('statusBox').textContent==='Agent status loaded.');
|
||||
await page.locator('#discoveriesBody tr').nth(1).getByRole('button',{name:'Enrich / Match',exact:true}).click();
|
||||
assert.equal(commands.length,2);
|
||||
assert.equal(commands[0].body.discovery_id,observations[0].id);
|
||||
assert.equal(commands[0].body.version,observations[0].version);
|
||||
assert.equal(commands[1].body.discovery_id,observations[1].id);
|
||||
assert.equal(commands[1].body.version,observations[1].version);
|
||||
assert.ok(commands.every(c=>!Object.hasOwn(c.body,'ip_address')));
|
||||
assert.deepEqual(unexpected,[]);
|
||||
console.log('Chromium passed: stored markup inert; Agent/status/error data escaped; same-IP actions retain exact observation IDs/versions; no external requests.');
|
||||
}
|
||||
} finally {await browser.close();}
|
||||
})().catch(e=>{console.error(e);process.exitCode=1});
|
||||
@@ -0,0 +1,20 @@
|
||||
"""Export the literal production page without importing/startup or network access."""
|
||||
|
||||
import ast, json, sys
|
||||
from pathlib import Path
|
||||
|
||||
root = Path(sys.argv[1])
|
||||
tree = ast.parse((root / "backend/app/main.py").read_text())
|
||||
node = next(
|
||||
n
|
||||
for n in tree.body
|
||||
if isinstance(n, ast.AsyncFunctionDef) and n.name == "minerops_agent_control_page"
|
||||
)
|
||||
html = next(
|
||||
n.value
|
||||
for n in ast.walk(node)
|
||||
if isinstance(n, ast.Constant)
|
||||
and isinstance(n.value, str)
|
||||
and "<!doctype html>" in n.value
|
||||
)
|
||||
Path(sys.argv[2]).write_text(html)
|
||||
@@ -16,7 +16,7 @@ from app.agent_commands.exceptions import (
|
||||
AgentCommandConflict,
|
||||
AgentResultValidationError,
|
||||
)
|
||||
from app.agent_commands.models import AgentCommandResultRequest
|
||||
from app.agent_commands.models import AgentCommandResultRequest, MinerControlResultPayload
|
||||
from app.agent_commands.repository import CLAIM_COMMANDS_SQL
|
||||
from app.agent_commands.router import router
|
||||
from app.agent_commands.service import AgentCommandService
|
||||
@@ -38,14 +38,14 @@ class FakeRepository:
|
||||
def __init__(self, row=None):
|
||||
self.row = row or {
|
||||
"id": "command-1", "agent_id": "test-agent",
|
||||
"command_type": "miner_control", "command_status": "claimed",
|
||||
"command_type": "poll_known_miners_now", "command_status": "claimed",
|
||||
"request_payload": {"action": "suspend"},
|
||||
}
|
||||
self.finished = []
|
||||
|
||||
def claim_pending(self, agent_id, limit=5):
|
||||
return [{
|
||||
"command_id": "command-1", "command_type": "miner_control",
|
||||
"command_id": "command-1", "command_type": "poll_known_miners_now",
|
||||
"request_payload": {"action": "suspend"}, "created_at": NOW,
|
||||
}]
|
||||
|
||||
@@ -54,6 +54,8 @@ class FakeRepository:
|
||||
|
||||
def finish_command(self, *args):
|
||||
self.finished.append(args)
|
||||
if args[2] == "cancelled":
|
||||
raise AgentCommandConflict(args[4])
|
||||
|
||||
|
||||
def typed_result(**overrides):
|
||||
@@ -84,7 +86,7 @@ class AgentCommandServiceTests(unittest.TestCase):
|
||||
)
|
||||
AgentCommandService(repository).submit_result("test-agent", request)
|
||||
finished = repository.finished[0]
|
||||
self.assertIsNone(finished[4])
|
||||
self.assertIn("[address]", finished[4])
|
||||
self.assertNotIn("192.0.2.10", str(finished))
|
||||
|
||||
def test_wrong_agent_pending_and_terminal_results_are_rejected(self):
|
||||
@@ -111,8 +113,8 @@ class AgentCommandServiceTests(unittest.TestCase):
|
||||
command_id="command-1", status="completed",
|
||||
result_payload={**typed_result(), forbidden: "sensitive"},
|
||||
)
|
||||
with self.assertRaises(AgentResultValidationError):
|
||||
AgentCommandService(repository).submit_result("test-agent", request)
|
||||
with self.assertRaises(ValueError):
|
||||
MinerControlResultPayload.model_validate(request.result_payload)
|
||||
|
||||
def test_suspend_resume_cannot_claim_confirmation(self):
|
||||
for status in ("confirmed", "verification_failed"):
|
||||
@@ -123,8 +125,10 @@ class AgentCommandServiceTests(unittest.TestCase):
|
||||
verification_attempted=True, verification_status=status
|
||||
),
|
||||
)
|
||||
with self.assertRaises(AgentResultValidationError):
|
||||
AgentCommandService(FakeRepository()).submit_result("test-agent", request)
|
||||
repository = FakeRepository()
|
||||
repository.row["command_type"] = "miner_control"
|
||||
with self.assertRaisesRegex(AgentCommandConflict, "physical_control_disabled_stage_a"):
|
||||
AgentCommandService(repository).submit_result("test-agent", request)
|
||||
|
||||
|
||||
class FakeAgentAuthRepository:
|
||||
@@ -160,17 +164,26 @@ class AgentCommandRouterTests(unittest.TestCase):
|
||||
"X-MinerOps-Agent-Secret": "test-secret",
|
||||
}
|
||||
|
||||
def test_poll_requires_header_identity_and_ignores_query_spoofing(self):
|
||||
def test_poll_accepts_secret_only_legacy_agent_and_ignores_query_spoofing(self):
|
||||
before_claim_calls = []
|
||||
self.app.state.agent_command_before_claim = lambda: before_claim_calls.append(True)
|
||||
self.assertEqual(self.client.get("/api/agent/commands/pending").status_code, 401)
|
||||
self.assertEqual(
|
||||
self.client.get(
|
||||
"/api/agent/commands/pending",
|
||||
headers={"X-MinerOps-Agent-Secret": "test-secret"},
|
||||
).status_code,
|
||||
200,
|
||||
)
|
||||
self.assertEqual(
|
||||
self.client.get(
|
||||
"/api/agent/commands/pending?agent_id=other", headers=self.headers()
|
||||
).status_code,
|
||||
200,
|
||||
)
|
||||
self.assertEqual(before_claim_calls, [True])
|
||||
# Agent polling may claim already queued commands, but must never invoke
|
||||
# a legacy scheduler/materializer as a side effect.
|
||||
self.assertEqual(before_claim_calls, [])
|
||||
self.assertEqual(
|
||||
self.client.get(
|
||||
"/api/agent/commands/pending", headers=self.headers("other")
|
||||
@@ -214,9 +227,9 @@ class AgentSourceContractTests(unittest.TestCase):
|
||||
self.assertIn('.agent_zone_id', service)
|
||||
self.assertIn('"agent_zone_id"', (root / "config.example.json").read_text())
|
||||
self.assertNotIn("?agent_id=", service)
|
||||
self.assertIn('"get.miner.setting"', control)
|
||||
self.assertIn("$attempt -le 3", control)
|
||||
self.assertIn("$Payload.performance_mode", control)
|
||||
self.assertIn("physical_control_disabled_stage_a", control)
|
||||
self.assertNotIn('"set.miner.', control)
|
||||
self.assertNotIn("$Payload.performance_mode", control)
|
||||
self.assertNotIn("api3_account =", control)
|
||||
self.assertNotIn("api3_password =", control)
|
||||
for forbidden in ("raw_response", "raw_command", "raw_exception"):
|
||||
@@ -251,30 +264,13 @@ if __name__ == "__main__":
|
||||
|
||||
class AgentTransportIsolationSourceTests(unittest.TestCase):
|
||||
def test_known_miners_and_pool_snapshots_are_agent_zone_scoped(self):
|
||||
main = (Path(__file__).parents[1] / "app" / "main.py").read_text()
|
||||
identity = (Path(__file__).parents[1] / "app" / "identity.py").read_text()
|
||||
|
||||
known_start = main.index('@app.get("/api/agent/known-miners")')
|
||||
known_end = main.index("# Phase 5J:", known_start)
|
||||
known = main[known_start:known_end]
|
||||
|
||||
pool_start = main.index('@app.post("/api/agent/pool-snapshots")')
|
||||
pool_end = main.index("\n@app.", pool_start + 1)
|
||||
pool = main[pool_start:pool_end]
|
||||
|
||||
self.assertIn('"/api/agent/known-miners"', identity)
|
||||
|
||||
self.assertIn("Depends(require_agent_identity)", known)
|
||||
self.assertIn("zone_service.routing_context", known)
|
||||
self.assertIn("miners.agent_zone_id = $1::uuid", known)
|
||||
self.assertNotIn("minerops_require_agent_secret", known)
|
||||
|
||||
self.assertIn("Depends(require_agent_identity)", pool)
|
||||
self.assertIn("zone_service.routing_context", pool)
|
||||
self.assertIn("agent_id = authenticated_agent_id", pool)
|
||||
self.assertIn("agent_zone_id = %s", pool)
|
||||
self.assertNotIn('body.get("agent_id")', pool)
|
||||
self.assertNotIn("MINEROPS_AGENT_SHARED_SECRET", pool)
|
||||
from app.main import minerops_agent_known_miners, agent_pool_snapshots
|
||||
from inspect import getsource
|
||||
for fn,method in [(minerops_agent_known_miners,"known"),(agent_pool_snapshots,"snapshots")]:
|
||||
source=getsource(fn)
|
||||
self.assertIn("Depends(require_agent_identity)",source)
|
||||
self.assertIn("discovery_dispatch",source)
|
||||
self.assertIn("agent_id=authenticated_agent_id",source)
|
||||
|
||||
def test_windows_machine_pollers_send_agent_identity(self):
|
||||
root = agent_source_root()
|
||||
@@ -294,67 +290,21 @@ class AgentTransportIsolationSourceTests(unittest.TestCase):
|
||||
|
||||
class AgentEnrichmentIsolationSourceTests(unittest.TestCase):
|
||||
def test_enrichment_routes_require_agent_identity_and_zone_scope(self):
|
||||
main = (Path(__file__).parents[1] / "app" / "main.py").read_text()
|
||||
|
||||
decorators = (
|
||||
'@app.post("/api/agent/miner-enrichment")',
|
||||
'@app.post("/api/agent/miner-enrichment-v2")',
|
||||
'@app.post("/api/agent/miner-enrichment-v3")',
|
||||
)
|
||||
|
||||
for decorator in decorators:
|
||||
with self.subTest(decorator=decorator):
|
||||
start = main.index(decorator)
|
||||
end = main.index("\n@app.", start + 1)
|
||||
route = main[start:end]
|
||||
|
||||
self.assertIn("Depends(require_agent_identity)", route)
|
||||
self.assertIn("zone_service.routing_context", route)
|
||||
self.assertIn("agent_zone_id", route)
|
||||
self.assertNotIn("minerops_require_agent_secret", route)
|
||||
|
||||
v2_start = main.index('@app.post("/api/agent/miner-enrichment-v2")')
|
||||
v2_end = main.index("\n@app.", v2_start + 1)
|
||||
v2 = main[v2_start:v2_end]
|
||||
self.assertIn(
|
||||
"AND agent_zone_id = $2::uuid",
|
||||
v2,
|
||||
)
|
||||
self.assertIn(
|
||||
"AND agent_zone_id = $3::uuid",
|
||||
v2,
|
||||
)
|
||||
|
||||
v3_start = main.index('@app.post("/api/agent/miner-enrichment-v3")')
|
||||
v3_end = main.index("\n@app.", v3_start + 1)
|
||||
v3 = main[v3_start:v3_end]
|
||||
self.assertIn(
|
||||
"AND agent_zone_id = $8::uuid",
|
||||
v3,
|
||||
)
|
||||
self.assertIn(
|
||||
"AND agent_zone_id = $7::uuid",
|
||||
v3,
|
||||
)
|
||||
from app import main
|
||||
from inspect import getsource
|
||||
for fn in [main.minerops_agent_miner_enrichment,main.minerops_agent_miner_enrichment_v2,main.minerops_agent_miner_enrichment_v3]:
|
||||
source=getsource(fn)
|
||||
self.assertIn("Depends(require_agent_identity)",source)
|
||||
self.assertIn("discovery_dispatch",source)
|
||||
self.assertIn("agent_id=authenticated_agent_id",source)
|
||||
|
||||
def test_api3_uses_authenticated_identity_and_never_global_agent_identity(self):
|
||||
main = (Path(__file__).parents[1] / "app" / "main.py").read_text()
|
||||
|
||||
start = main.index('@app.post("/api/agent/api3-enrichment")')
|
||||
end = main.index("\n@app.", start + 1)
|
||||
route = main[start:end]
|
||||
|
||||
self.assertIn("Depends(require_agent_identity)", route)
|
||||
self.assertIn("zone_service.routing_context", route)
|
||||
self.assertIn("agent_id = authenticated_agent_id", route)
|
||||
self.assertIn("agent_zone_id = %s", route)
|
||||
self.assertIn("if miner_id is None:", route)
|
||||
self.assertIn("continue", route)
|
||||
|
||||
self.assertNotIn("MINEROPS_AGENT_SHARED_SECRET", route)
|
||||
self.assertNotIn('body.get("agent_id")', route)
|
||||
self.assertNotIn('body.get("site_name")', route)
|
||||
self.assertNotIn('request.headers.get("x-agent-secret")', route)
|
||||
from app.main import minerops_agent_api3_enrichment
|
||||
from inspect import getsource
|
||||
source=getsource(minerops_agent_api3_enrichment)
|
||||
self.assertIn("Depends(require_agent_identity)",source)
|
||||
self.assertIn("discovery_dispatch",source)
|
||||
self.assertIn("agent_id=authenticated_agent_id",source)
|
||||
|
||||
def test_legacy_config_remains_explicitly_legacy_only(self):
|
||||
main = (Path(__file__).parents[1] / "app" / "main.py").read_text()
|
||||
@@ -366,3 +316,18 @@ class AgentEnrichmentIsolationSourceTests(unittest.TestCase):
|
||||
self.assertIn("minerops_require_agent_secret(request)", route)
|
||||
self.assertIn('"site_name": "Plexus"', route)
|
||||
self.assertNotIn("Depends(require_agent_identity)", route)
|
||||
|
||||
def test_disabled_old_command_routes_are_fixed_retirement_responses(self):
|
||||
main = (Path(__file__).parents[1] / "app" / "main.py").read_text()
|
||||
for decorator in (
|
||||
'@app.get("/api/agent/commands/pending-disabled-old")',
|
||||
'@app.post("/api/agent/commands/result-disabled-old")',
|
||||
):
|
||||
with self.subTest(decorator=decorator):
|
||||
start = main.index(decorator)
|
||||
end = main.index("\n@app.", start + 1)
|
||||
route = main[start:end]
|
||||
self.assertIn("status_code=410", route)
|
||||
self.assertNotIn("minerops_require_agent_secret", route)
|
||||
self.assertNotIn("request.json", route)
|
||||
self.assertNotIn("minerops_agent_command_queue", route)
|
||||
@@ -0,0 +1,39 @@
|
||||
"""Reused discovery isolation scenarios exercised through real HTTP authority."""
|
||||
|
||||
import unittest
|
||||
from tests.test_discovery_security_postgres import DiscoveryFixtures
|
||||
|
||||
|
||||
class AgentDiscoveryIsolationPostgresTests(DiscoveryFixtures, unittest.TestCase):
|
||||
def test_cross_zone_mac_and_ip_are_unmatched_without_id_disclosure(self):
|
||||
mid, mac = self.miner(1)
|
||||
before = self.durable()
|
||||
row = self.observe(mac=mac)
|
||||
self.assertEqual(row["discovery_status"], "zone_conflict")
|
||||
self.assertIsNone(row["matched_miner_id"])
|
||||
self.assertNotIn(str(mid), str(row))
|
||||
self.assertEqual(before["miners"], self.durable()["miners"])
|
||||
|
||||
def test_same_zone_mac_and_ip_matches_are_preserved(self):
|
||||
mid, mac = self.miner()
|
||||
row = self.observe(mac=mac)
|
||||
self.assertEqual(row["matched_miner_id"], str(mid))
|
||||
self.assertEqual(row["discovery_status"], "known")
|
||||
|
||||
def test_unassigned_miner_is_not_claimed_by_discovery_matching(self):
|
||||
mid, mac = self.miner()
|
||||
with __import__(
|
||||
"app.agent_zones.repository", fromlist=["default_connection_factory"]
|
||||
).default_connection_factory() as conn:
|
||||
conn.execute("UPDATE miners SET agent_zone_id=NULL WHERE id=%s", (mid,))
|
||||
row = self.observe(mac=mac)
|
||||
self.assertIsNone(row["matched_miner_id"])
|
||||
self.assertIsNone(
|
||||
self.scalar("SELECT agent_zone_id FROM miners WHERE id=%s", (mid,))
|
||||
)
|
||||
|
||||
def test_claimed_agent_identity_cannot_override_authentication(self):
|
||||
response = self.post(
|
||||
"/api/agent/discoveries", {"agent_id": self.agents[1], "discoveries": []}
|
||||
)
|
||||
self.assertEqual(response.status_code, 403)
|
||||
@@ -42,6 +42,7 @@ class FakeRepository:
|
||||
self.agent = None
|
||||
self.saved = []
|
||||
self.assign_result = {MINER_A: "assigned"}
|
||||
self.discovery_match = None
|
||||
|
||||
def get_agent(self, agent_id):
|
||||
return self.agent if self.agent and self.agent["agent_id"] == agent_id else None
|
||||
@@ -107,6 +108,11 @@ class FakeRepository:
|
||||
],
|
||||
}
|
||||
|
||||
def find_miner_in_zone(self, *, agent_id, agent_zone_id, mac_address, ip_address):
|
||||
self.discovery_match_request = {'agent_id':agent_id,'agent_zone_id':agent_zone_id,
|
||||
'mac_address':mac_address,'ip_address':ip_address}
|
||||
return self.discovery_match
|
||||
|
||||
def assign_miners_to_zone(self, miner_ids, agent_zone_id):
|
||||
self.assigned = (list(miner_ids), agent_zone_id)
|
||||
return {miner_id: self.assign_result.get(miner_id, "miner_not_found")
|
||||
@@ -136,6 +142,7 @@ class AgentZoneModelTests(unittest.TestCase):
|
||||
class AgentZoneServiceTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.repository = FakeRepository()
|
||||
self.repository.agent = {"agent_id":"agent-a","site_id":SITE_A,"agent_zone_id":ZONE_A,"site_name":"North"}
|
||||
self.service = AgentZoneService(self.repository)
|
||||
|
||||
def checkin(self, **kwargs):
|
||||
@@ -144,14 +151,14 @@ class AgentZoneServiceTests(unittest.TestCase):
|
||||
AgentCheckinRequest(agent_id="agent-a", site_name="North", **kwargs),
|
||||
)
|
||||
|
||||
def test_explicit_first_assignment_and_same_assignment_are_allowed(self):
|
||||
def test_established_assignment_and_same_assignment_are_allowed(self):
|
||||
first = self.checkin(site_id=SITE_A, agent_zone_id=ZONE_A)
|
||||
second = self.checkin(site_id=SITE_A, agent_zone_id=ZONE_A)
|
||||
self.assertTrue(first.routing_available)
|
||||
self.assertEqual(second.agent_zone_id, ZONE_A)
|
||||
|
||||
def test_explicit_zone_site_mismatch_and_identity_spoof_are_rejected(self):
|
||||
with self.assertRaisesRegex(AgentZoneValidationError, "does not belong"):
|
||||
with self.assertRaisesRegex(AgentZoneValidationError, "cannot change Sites"):
|
||||
self.checkin(site_id=SITE_B, agent_zone_id=ZONE_A)
|
||||
with self.assertRaisesRegex(AgentZoneValidationError, "authenticated"):
|
||||
self.service.checkin(
|
||||
@@ -181,34 +188,20 @@ class AgentZoneServiceTests(unittest.TestCase):
|
||||
self.assertEqual(result.agent_zone_id, ZONE_A)
|
||||
|
||||
def test_transitional_resolution_requires_exactly_one_enabled_zone(self):
|
||||
single = self.checkin(site_id=SITE_A)
|
||||
self.assertEqual(single.agent_zone_id, ZONE_A)
|
||||
self.repository.agent = None
|
||||
self.repository.zones = {}
|
||||
zero = self.checkin(site_id=SITE_A)
|
||||
self.assertEqual(zero.routing_reason, "agent_zone_unavailable")
|
||||
self.repository.agent = None
|
||||
self.repository.zones = {
|
||||
ZONE_A: zone(),
|
||||
ZONE_B: zone(ZONE_B, SITE_A, name="Room 2"),
|
||||
}
|
||||
many = self.checkin(site_id=SITE_A)
|
||||
self.assertEqual(many.routing_reason, "agent_zone_ambiguous")
|
||||
self.assertIsNone(many.agent_zone_id)
|
||||
for zones in [{}, {ZONE_A: zone()}, {ZONE_A:zone(),ZONE_B:zone(ZONE_B,SITE_A)}]:
|
||||
self.repository.zones=zones
|
||||
self.repository.agent=None
|
||||
with self.assertRaisesRegex(AgentZoneValidationError,"administrator-established"):
|
||||
self.checkin(site_id=SITE_A)
|
||||
|
||||
def test_disabled_zone_is_not_selected_and_existing_assignment_is_factual(self):
|
||||
self.repository.zones[ZONE_A]["is_enabled"] = False
|
||||
unassigned = self.checkin(site_id=SITE_A)
|
||||
self.assertEqual(unassigned.routing_reason, "agent_zone_unavailable")
|
||||
self.repository.agent = {
|
||||
"agent_id": "agent-a", "site_id": SITE_A,
|
||||
"agent_zone_id": ZONE_A, "site_name": "North",
|
||||
}
|
||||
existing = self.service.checkin(
|
||||
"agent-a", AgentCheckinRequest(agent_id="agent-a")
|
||||
)
|
||||
self.assertEqual(existing.routing_reason, "agent_zone_disabled")
|
||||
self.assertEqual(existing.agent_zone_id, ZONE_A)
|
||||
self.repository.zones[ZONE_A]["is_enabled"]=False
|
||||
existing=self.service.checkin("agent-a",AgentCheckinRequest(agent_id="agent-a"))
|
||||
self.assertEqual(existing.routing_reason,"agent_zone_disabled")
|
||||
self.assertEqual(existing.agent_zone_id,ZONE_A)
|
||||
self.repository.agent=None
|
||||
with self.assertRaisesRegex(AgentZoneValidationError,"administrator-established"):
|
||||
self.checkin(site_id=SITE_A)
|
||||
|
||||
def test_payload_is_sanitized_before_persistence(self):
|
||||
request = AgentCheckinRequest.model_validate({
|
||||
@@ -255,9 +248,41 @@ class AgentZoneServiceTests(unittest.TestCase):
|
||||
self.assertNotIn("10.80.0.0/24", str(configuration))
|
||||
|
||||
|
||||
def test_discovery_match_uses_only_authenticated_agent_zone(self):
|
||||
self.checkin(site_id=SITE_A, agent_zone_id=ZONE_A)
|
||||
self.repository.discovery_match = {"id": MINER_A, "matched_by": "mac"}
|
||||
|
||||
result = self.service.match_discovered_miner(
|
||||
"agent-a",
|
||||
mac_address="02:00:00:00:16:21",
|
||||
ip_address="192.0.2.21",
|
||||
)
|
||||
|
||||
self.assertEqual(result.miner_id, MINER_A)
|
||||
self.assertEqual(result.matched_by, "mac")
|
||||
self.assertEqual(
|
||||
self.repository.discovery_match_request["agent_zone_id"], ZONE_A
|
||||
)
|
||||
|
||||
def test_discovery_match_rejects_stale_recorded_zone_without_querying(self):
|
||||
self.checkin(site_id=SITE_A, agent_zone_id=ZONE_A)
|
||||
|
||||
result = self.service.match_discovered_miner(
|
||||
"agent-a",
|
||||
recorded_zone_id=ZONE_B,
|
||||
mac_address="02:00:00:00:16:21",
|
||||
ip_address="192.0.2.21",
|
||||
)
|
||||
|
||||
self.assertIsNone(result.miner_id)
|
||||
self.assertEqual(result.routing_reason, "agent_zone_conflict")
|
||||
self.assertFalse(hasattr(self.repository, "discovery_match_request"))
|
||||
|
||||
|
||||
class AgentZoneRouterTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.repository = FakeRepository()
|
||||
self.repository.agent = {"agent_id":"agent-a","site_id":SITE_A,"agent_zone_id":ZONE_A,"site_name":"North"}
|
||||
self.service = AgentZoneService(self.repository)
|
||||
self.app = FastAPI()
|
||||
self.app.include_router(router)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import unittest
|
||||
from uuid import UUID
|
||||
from unittest.mock import patch
|
||||
|
||||
import psycopg
|
||||
from psycopg.errors import ForeignKeyViolation, UniqueViolation
|
||||
@@ -10,6 +11,8 @@ from app.agent_zones.repository import AgentZoneRepository
|
||||
from app.config import settings
|
||||
from app.miner_control.repository import MinerControlRepository
|
||||
from app.miner_control.service import MinerControlService
|
||||
from app.miner_control.exceptions import MinerControlUnavailable
|
||||
from app.agent_commands.policy import PHYSICAL_CONTROL_DISABLED
|
||||
|
||||
|
||||
SITE_A = UUID("00000000-0000-4000-8000-000000001630")
|
||||
@@ -34,6 +37,29 @@ def connect():
|
||||
|
||||
|
||||
class AgentZonePostgresTests(unittest.TestCase):
|
||||
def assert_stage_a_unavailable(self, expected_zone, expected_agent):
|
||||
repository = MinerControlRepository(connect)
|
||||
context = repository.get_control_context(MINER_UNASSIGNED)
|
||||
# Check actual routing facts independently of the unconditional Stage A
|
||||
# reason. A changed assertion must not hide a Site fallback regression.
|
||||
self.assertEqual(context["agent_zone_id"], expected_zone)
|
||||
self.assertEqual(context["agent_id"], expected_agent)
|
||||
projection = MinerControlService(repository).get_controls(MINER_UNASSIGNED)
|
||||
for capability in (projection.service_control, projection.performance_mode_control):
|
||||
self.assertEqual(capability.state, "unavailable")
|
||||
self.assertEqual(capability.reason, PHYSICAL_CONTROL_DISABLED)
|
||||
self.assertIsNone(projection.available_service_action)
|
||||
with patch.object(repository, "enqueue_with_audit") as enqueue:
|
||||
for action, value in (("suspend", None), ("resume", None), ("performance_mode", "high")):
|
||||
with self.assertRaisesRegex(MinerControlUnavailable, PHYSICAL_CONTROL_DISABLED):
|
||||
MinerControlService(repository).queue_control(
|
||||
MINER_UNASSIGNED, action, requested_value=value,
|
||||
actor_user_id=UUID("00000000-0000-4000-8000-000000001639"),
|
||||
actor_username="zone-containment-test",
|
||||
)
|
||||
enqueue.assert_not_called()
|
||||
return context
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
connection = connect()
|
||||
@@ -170,12 +196,12 @@ class AgentZonePostgresTests(unittest.TestCase):
|
||||
connection.rollback()
|
||||
connection.close()
|
||||
|
||||
def test_unassigned_miner_inherits_zone_and_existing_other_zone_wins(self):
|
||||
def test_unassigned_miner_stays_unassigned_and_existing_other_zone_wins(self):
|
||||
repository = AgentZoneRepository(connect)
|
||||
outcomes = repository.assign_miners_to_zone(
|
||||
[MINER_UNASSIGNED, MINER_OTHER], ZONE_A
|
||||
)
|
||||
self.assertEqual(outcomes[MINER_UNASSIGNED], "assigned")
|
||||
self.assertEqual(outcomes[MINER_UNASSIGNED], "miner_zone_unassigned")
|
||||
self.assertEqual(outcomes[MINER_OTHER], "agent_zone_conflict")
|
||||
connection = connect()
|
||||
try:
|
||||
@@ -187,7 +213,7 @@ class AgentZonePostgresTests(unittest.TestCase):
|
||||
assignments = dict(cur.fetchall())
|
||||
finally:
|
||||
connection.close()
|
||||
self.assertEqual(assignments[MINER_UNASSIGNED], ZONE_A)
|
||||
self.assertIsNone(assignments[MINER_UNASSIGNED])
|
||||
self.assertEqual(assignments[MINER_OTHER], ZONE_B)
|
||||
|
||||
def test_disabled_zone_blocks_control_without_site_fallback(self):
|
||||
@@ -203,10 +229,8 @@ class AgentZonePostgresTests(unittest.TestCase):
|
||||
finally:
|
||||
connection.close()
|
||||
try:
|
||||
projection = MinerControlService(MinerControlRepository(connect)).get_controls(
|
||||
MINER_UNASSIGNED
|
||||
)
|
||||
self.assertEqual(projection.service_control.reason, "agent_zone_disabled")
|
||||
context = self.assert_stage_a_unavailable(ZONE_A, AGENT_A)
|
||||
self.assertFalse(context["agent_zone_enabled"])
|
||||
finally:
|
||||
connection = connect()
|
||||
try:
|
||||
@@ -228,16 +252,10 @@ class AgentZonePostgresTests(unittest.TestCase):
|
||||
connection.commit()
|
||||
finally:
|
||||
connection.close()
|
||||
projection = MinerControlService(MinerControlRepository(connect)).get_controls(
|
||||
MINER_UNASSIGNED
|
||||
)
|
||||
self.assertEqual(projection.service_control.reason, "agent_unavailable")
|
||||
self.assert_stage_a_unavailable(ZONE_C, None)
|
||||
|
||||
def test_null_zone_does_not_fall_back_to_site_agent(self):
|
||||
projection = MinerControlService(MinerControlRepository(connect)).get_controls(
|
||||
MINER_UNASSIGNED
|
||||
)
|
||||
self.assertEqual(projection.service_control.reason, "agent_zone_unassigned")
|
||||
self.assert_stage_a_unavailable(None, None)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,243 @@
|
||||
"""Focused regression tests for retired legacy miner-write surfaces."""
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import main
|
||||
from app.identity import get_current_user, require_permission, required_api_permission
|
||||
|
||||
|
||||
APP_ROOT = Path(__file__).parents[1] / "app"
|
||||
|
||||
|
||||
class LegacyControlRetirementTests(unittest.TestCase):
|
||||
@staticmethod
|
||||
def assert_retired(callable_, *args, **kwargs):
|
||||
try:
|
||||
result = callable_(*args, **kwargs)
|
||||
if inspect.isawaitable(result):
|
||||
asyncio.run(result)
|
||||
except HTTPException as exc:
|
||||
if exc.status_code != 410:
|
||||
raise AssertionError(f"Expected HTTP 410, got {exc.status_code}") from exc
|
||||
return
|
||||
raise AssertionError("Expected a fixed HTTP 410 retirement response")
|
||||
|
||||
def test_all_legacy_http_write_routes_are_fixed_inert_responses(self):
|
||||
routes = (
|
||||
(main.run_dashboard_m30_command_test_lab, {"role": "Admin"}),
|
||||
(main.retired_direct_reboot_miner_command, "miner-1", {"role": "Admin"}),
|
||||
(main.real_power_off_miner_command, "miner-1", {"role": "Admin"}),
|
||||
(main.real_power_on_miner_command, "miner-1", {"role": "Admin"}),
|
||||
(main.real_suspend_miner_command, "miner-1", {"role": "Admin"}),
|
||||
(main.real_resume_miner_command, "miner-1", {"role": "Admin"}),
|
||||
(main.retired_direct_api3_password_change, "miner-1", {"role": "Admin"}),
|
||||
(main.retired_legacy_agent_miner_control, "miner-1", {"role": "Admin"}),
|
||||
(main.process_due_dashboard_scheduled_bulk_actions, {"role": "Admin"}),
|
||||
(main.create_dashboard_scheduled_bulk_action, {"role": "Admin"}),
|
||||
(main.whatsminer_read_test, {"role": "Admin"}),
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(main, "get_db_connection") as sync_database,
|
||||
patch.object(main, "minerops_command_db_connect") as async_database,
|
||||
patch.object(main, "insert_miner_command_event") as audit_write,
|
||||
patch.object(main, "minerops_send_whatsminer_reboot") as reboot_write,
|
||||
patch.object(main, "minerops_power_command") as power_write,
|
||||
patch.object(main, "minerops_api3_change_password") as password_write,
|
||||
):
|
||||
for route in routes:
|
||||
with self.subTest(route=route[0].__name__):
|
||||
self.assert_retired(route[0], *route[1:])
|
||||
|
||||
for forbidden_call in (
|
||||
sync_database,
|
||||
async_database,
|
||||
audit_write,
|
||||
reboot_write,
|
||||
power_write,
|
||||
password_write,
|
||||
):
|
||||
self.assertFalse(forbidden_call.called)
|
||||
|
||||
def test_m30_requires_control_permission_before_inert_response(self):
|
||||
self.assertEqual(
|
||||
required_api_permission(
|
||||
"POST", "/api/dashboard/m30-command-test-lab/run"
|
||||
),
|
||||
"miners.control",
|
||||
)
|
||||
dependency = require_permission("miners.control")
|
||||
with self.assertRaises(HTTPException) as denied:
|
||||
dependency({"role": "Read Only", "permissions": []})
|
||||
self.assertEqual(denied.exception.status_code, 403)
|
||||
|
||||
route_source = inspect.getsource(main.run_dashboard_m30_command_test_lab)
|
||||
self.assertIn('Depends(require_permission("miners.control"))', route_source)
|
||||
self.assertIn("status_code=410", route_source)
|
||||
self.assertNotIn("minerops_agent_command_queue", route_source)
|
||||
self.assertNotIn("plexus-windows-agent", route_source)
|
||||
self.assertNotIn("dry_run", route_source)
|
||||
|
||||
app = FastAPI()
|
||||
app.add_api_route(
|
||||
"/api/dashboard/m30-command-test-lab/run",
|
||||
main.run_dashboard_m30_command_test_lab,
|
||||
methods=["POST"],
|
||||
)
|
||||
with TestClient(app) as client:
|
||||
app.dependency_overrides[get_current_user] = lambda: {
|
||||
"role": "Read Only",
|
||||
"permissions": [],
|
||||
}
|
||||
self.assertEqual(
|
||||
client.post("/api/dashboard/m30-command-test-lab/run").status_code,
|
||||
403,
|
||||
)
|
||||
app.dependency_overrides[get_current_user] = lambda: {
|
||||
"role": "Custom",
|
||||
"permissions": ["miners.control"],
|
||||
}
|
||||
self.assertEqual(
|
||||
client.post("/api/dashboard/m30-command-test-lab/run").status_code,
|
||||
410,
|
||||
)
|
||||
|
||||
def test_registered_retired_routes_have_no_direct_write_implementation(self):
|
||||
retired_paths = {
|
||||
"/api/dashboard/m30-command-test-lab/run",
|
||||
"/api/dashboard/scheduled-bulk-actions/process-due",
|
||||
"/api/dashboard/scheduled-bulk-actions",
|
||||
"/api/tools/whatsminer-read-test",
|
||||
"/api/miners/{miner_id}/commands/reboot",
|
||||
"/api/miners/{miner_id}/commands/power-off",
|
||||
"/api/miners/{miner_id}/commands/power-on",
|
||||
"/api/miners/{miner_id}/commands/suspend",
|
||||
"/api/miners/{miner_id}/commands/resume",
|
||||
"/api/miners/{miner_id}/api3/change-password",
|
||||
"/api/miners/{miner_id}/agent-control",
|
||||
}
|
||||
registered = [
|
||||
(route.path, route.endpoint)
|
||||
for route in main.app.routes
|
||||
if getattr(route, "path", None) in retired_paths
|
||||
and "POST" in getattr(route, "methods", set())
|
||||
]
|
||||
self.assertEqual({path for path, _ in registered}, retired_paths)
|
||||
self.assertEqual(len(registered), len(retired_paths))
|
||||
|
||||
forbidden = (
|
||||
"socket",
|
||||
"minerops_send_",
|
||||
"minerops_power_command",
|
||||
"minerops_api3_change_password",
|
||||
"insert_miner_command_event",
|
||||
"minerops_agent_command_queue",
|
||||
"minerops_command_db_connect",
|
||||
"get_db_connection",
|
||||
"request.json",
|
||||
)
|
||||
for path, endpoint in registered:
|
||||
source = inspect.getsource(endpoint)
|
||||
with self.subTest(path=path):
|
||||
self.assertIn("status_code=410", source)
|
||||
for unsafe_fragment in forbidden:
|
||||
self.assertNotIn(unsafe_fragment, source)
|
||||
|
||||
def test_legacy_scheduler_is_noop_and_not_registered_before_claim(self):
|
||||
with patch.object(main, "get_db_connection") as database:
|
||||
self.assertEqual(main.minerops_process_due_scheduled_bulk_actions(), [])
|
||||
self.assertFalse(database.called)
|
||||
self.assertFalse(hasattr(main.app.state, "agent_command_before_claim"))
|
||||
source = inspect.getsource(main.minerops_process_due_scheduled_bulk_actions)
|
||||
self.assertNotIn("minerops_agent_command_queue", source)
|
||||
self.assertNotIn("get_db_connection", source)
|
||||
|
||||
def test_every_retired_http_route_denies_unprivileged_users_and_never_writes(self):
|
||||
paths = {
|
||||
"/api/dashboard/m30-command-test-lab/run",
|
||||
"/api/dashboard/scheduled-bulk-actions",
|
||||
"/api/dashboard/scheduled-bulk-actions/process-due",
|
||||
"/api/tools/whatsminer-read-test",
|
||||
"/api/miners/{miner_id}/commands/reboot",
|
||||
"/api/miners/{miner_id}/commands/power-off",
|
||||
"/api/miners/{miner_id}/commands/power-on",
|
||||
"/api/miners/{miner_id}/commands/suspend",
|
||||
"/api/miners/{miner_id}/commands/resume",
|
||||
"/api/miners/{miner_id}/api3/change-password",
|
||||
"/api/miners/{miner_id}/agent-control",
|
||||
}
|
||||
app = FastAPI()
|
||||
for route in main.app.routes:
|
||||
if getattr(route, "path", None) in paths and "POST" in route.methods:
|
||||
app.add_api_route(route.path, route.endpoint, methods=["POST"])
|
||||
with TestClient(app) as client, patch.object(main.socket, "create_connection") as transport, patch.object(main, "get_db_connection") as database:
|
||||
for permissions, expected in (([], 403), (["miners.control", "credentials.manage"], 410)):
|
||||
app.dependency_overrides[get_current_user] = lambda: {
|
||||
"role": "Custom", "permissions": permissions,
|
||||
}
|
||||
for path in paths:
|
||||
concrete = path.replace("{miner_id}", "miner-1")
|
||||
self.assertEqual(client.post(concrete, json={"dry_run": False}).status_code, expected, concrete)
|
||||
transport.assert_not_called()
|
||||
database.assert_not_called()
|
||||
|
||||
def test_diagnostic_write_disguises_overrides_and_api3_are_retired(self):
|
||||
app = FastAPI()
|
||||
app.add_api_route("/api/tools/whatsminer-read-test", main.whatsminer_read_test, methods=["POST"])
|
||||
app.dependency_overrides[get_current_user] = lambda: {"role": "Custom", "permissions": ["miners.control"]}
|
||||
attacks = (
|
||||
{"command": "reboot"}, {"command": "read_then_reboot"},
|
||||
{"command": "summary", "params": {"command": "reboot"}},
|
||||
{"command": "summary", "params": {"cmd": "set.miner.service", "param": "stop"}},
|
||||
{"port": 4433, "command": "set.miner.power_mode", "params": {"param": "high"}},
|
||||
{"command": "summary\n{\"command\":\"reboot\"}"},
|
||||
{"command": "summary"},
|
||||
)
|
||||
with TestClient(app) as client, patch.object(main.socket, "create_connection") as transport, patch.object(main, "send_whatsminer_read_command") as sender:
|
||||
for attack in attacks:
|
||||
self.assertEqual(client.post("/api/tools/whatsminer-read-test", json={"ip_address": "192.0.2.1", **attack}).status_code, 410)
|
||||
transport.assert_not_called()
|
||||
sender.assert_not_called()
|
||||
|
||||
def test_internal_discovery_helper_cannot_become_a_generic_command_gateway(self):
|
||||
with patch.object(main.socket, "create_connection") as transport:
|
||||
for command, port, params in (
|
||||
("reboot", 4028, None), ("read_reboot", 4028, None),
|
||||
("summary", 4433, None), ("set.miner.service", 4433, {"param": "stop"}),
|
||||
("summary", 4028, {"command": "reboot"}),
|
||||
("summary", 4028, {"cmd": "set.miner.service"}),
|
||||
("get_miner_info", 4028, {"info": "arbitrary"}),
|
||||
):
|
||||
self.assert_retired(main.send_whatsminer_read_command, "192.0.2.1", port, command, 500, params)
|
||||
transport.assert_not_called()
|
||||
|
||||
def test_old_agent_aliases_are_inert_even_without_authentication(self):
|
||||
app = FastAPI()
|
||||
app.add_api_route("/api/agent/commands/pending-disabled-old", main.minerops_agent_commands_pending_retired, methods=["GET"])
|
||||
app.add_api_route("/api/agent/commands/result-disabled-old", main.minerops_agent_commands_result_retired, methods=["POST"])
|
||||
with TestClient(app) as client, patch.object(main, "minerops_command_db_connect") as database:
|
||||
self.assertEqual(client.get("/api/agent/commands/pending-disabled-old").status_code, 410)
|
||||
self.assertEqual(client.post("/api/agent/commands/result-disabled-old", json={"status": "completed"}).status_code, 410)
|
||||
database.assert_not_called()
|
||||
|
||||
def test_canonical_control_still_derives_agent_from_miner_zone(self):
|
||||
repository_source = (
|
||||
APP_ROOT / "miner_control" / "repository.py"
|
||||
).read_text()
|
||||
self.assertIn("miners.agent_zone_id", repository_source)
|
||||
self.assertIn(
|
||||
"agents.agent_zone_id = miners.agent_zone_id",
|
||||
repository_source,
|
||||
)
|
||||
self.assertNotIn("agents.site_id = miners.site_id", repository_source)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,338 @@
|
||||
"""Audit F1-F3 reversals: real registered HTTP/transactions, no physical devices."""
|
||||
import ast
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
import os
|
||||
from pathlib import Path
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from uuid import uuid4
|
||||
|
||||
from fastapi import HTTPException
|
||||
import app.main as main
|
||||
from app.identity import create_access_token
|
||||
from tests.test_discovery_security_postgres import DiscoveryFixtures, connect
|
||||
|
||||
|
||||
class LegacyInventorySecurityTests(DiscoveryFixtures, unittest.TestCase):
|
||||
def upload(self, text, query='', headers=None):
|
||||
return self.client.post(
|
||||
'/api/imports/whatsminer-tool-scan?dry_run=false&update_existing=true' + query,
|
||||
headers=self.admin if headers is None else headers,
|
||||
files={'file': ('simulated.csv', text.encode(), 'text/csv')},
|
||||
)
|
||||
|
||||
def denied(self, response, before, status):
|
||||
self.assertEqual(response.status_code, status, response.text)
|
||||
self.assertEqual(before, self.durable())
|
||||
self.transport.assert_not_called()
|
||||
|
||||
def test_discovery_audit_reproductions_all_rejected_without_reads(self):
|
||||
self.miner(1, ip='198.51.100.111')
|
||||
self.miner(ip='198.51.100.125')
|
||||
self.miner(1, ip='198.51.100.125')
|
||||
for mode in ['true', 'false']:
|
||||
for ip in ['198.51.100.111', '198.51.100.125', '203.0.113.209']:
|
||||
with self.subTest(mode=mode, ip=ip), patch.dict(os.environ, MINEROPS_HUB_MODE=mode):
|
||||
before = self.durable()
|
||||
response = self.client.post('/api/miners/discover-one', headers=self.admin, json={'ip_address': ip})
|
||||
self.denied(response, before, 410)
|
||||
|
||||
def test_missing_disabled_agent_and_zone_cannot_restore_legacy_discovery(self):
|
||||
with connect() as c:
|
||||
c.execute('UPDATE agent_zones SET is_enabled=false WHERE id=%s', (self.zones[0],))
|
||||
c.execute('UPDATE minerops_site_agents SET credential_revoked_at=now() WHERE agent_id=%s', (self.agents[0],))
|
||||
c.execute('DELETE FROM minerops_site_agents WHERE agent_id=%s', (self.agents[1],))
|
||||
before = self.durable()
|
||||
r = self.client.post('/api/miners/discover-one', headers=self.admin, json={'ip_address': '198.51.100.111'})
|
||||
self.denied(r, before, 410)
|
||||
|
||||
def test_inventory_only_permission_cannot_replace_mac(self):
|
||||
role, uid = uuid4(), uuid4()
|
||||
with connect() as c:
|
||||
c.execute('INSERT INTO roles(id,name) VALUES (%s,%s)', (role, str(role)))
|
||||
c.execute("INSERT INTO role_permissions(role_id,permission_id) SELECT %s,id FROM permissions WHERE key='miners.update'", (role,))
|
||||
c.execute("INSERT INTO users(id,username,display_name,password_hash,role_id) VALUES (%s,%s,'Inventory only','unused',%s)", (uid, str(uid), role))
|
||||
headers = {'Authorization': 'Bearer ' + create_access_token(str(uid), str(uid), str(role))}
|
||||
mid, _ = self.miner(ip='198.51.100.127')
|
||||
before = self.durable()
|
||||
for path, body, status in [
|
||||
('/api/agent/discoveries/review', {'discovery_id': str(uuid4()), 'version': 'x', 'action': 'ignore'}, 403),
|
||||
(f'/api/miners/{mid}/replace-control-board', {}, 403),
|
||||
('/api/miners/discover-one', {'ip_address': '198.51.100.127'}, 410),
|
||||
]:
|
||||
self.denied(self.client.post(path, headers=headers, json=body), before, status)
|
||||
|
||||
def test_range_and_manual_scheduler_retired_in_both_modes(self):
|
||||
for mode in ['true', 'false']:
|
||||
with patch.dict(os.environ, MINEROPS_HUB_MODE=mode):
|
||||
for path in [f'/api/scan-ranges/{uuid4()}/discover-now', '/api/scanner/run-known-now']:
|
||||
before = self.durable()
|
||||
self.denied(self.client.post(path, headers=self.admin, json={}), before, 410)
|
||||
with patch('app.main.threading.Thread') as thread:
|
||||
main.start_scheduled_scanner()
|
||||
thread.assert_not_called()
|
||||
status = self.client.get('/api/scanner/status', headers=self.admin).json()['data']
|
||||
self.assertFalse(status['enabled'])
|
||||
self.assertTrue(status['scanner_disabled'])
|
||||
|
||||
def test_all_legacy_helper_definitions_and_saved_aliases_are_retired(self):
|
||||
# Execute every original definition, including shadowed functions; decorators
|
||||
# are omitted so this cannot mutate the application's registered routes.
|
||||
tree = ast.parse(Path(main.__file__).read_text())
|
||||
names = {'find_existing_miner', 'discover_one_miner', 'discover_scan_range_now',
|
||||
'call_discover_one_miner_flexible', 'refresh_one_known_miner_for_schedule',
|
||||
'run_scheduled_known_miner_scan', 'scheduled_scanner_loop',
|
||||
'apply_legacy_summary_to_miner', 'legacy_find_miner_id_by_ip'}
|
||||
import inspect
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.FunctionDef) and node.name in names:
|
||||
with self.subTest(helper=node.name, line=node.lineno):
|
||||
node.decorator_list = []
|
||||
namespace = dict(vars(main))
|
||||
exec(compile(ast.Module(body=[node], type_ignores=[]), '<retired-helper>', 'exec'), namespace)
|
||||
fn = namespace[node.name]
|
||||
kwargs = {name: None for name, p in inspect.signature(fn).parameters.items() if p.default is inspect.Parameter.empty}
|
||||
with patch('app.main.get_db_connection', side_effect=AssertionError('No legacy database access')):
|
||||
with self.assertRaises(HTTPException) as exc:
|
||||
fn(**kwargs)
|
||||
self.assertEqual(exc.exception.status_code, 410)
|
||||
with self.assertRaises(HTTPException) as exc:
|
||||
main._minerops_original_discover_one_miner_legacy_summary(None)
|
||||
self.assertEqual(exc.exception.status_code, 410)
|
||||
|
||||
def test_summary_id_ip_disagreement_and_bulk_retired(self):
|
||||
a, _ = self.miner(ip='198.51.100.121')
|
||||
self.miner(1, ip='198.51.100.122')
|
||||
with patch('app.main.read_legacy_whatsminer_summary', side_effect=AssertionError('No summary transport')) as read:
|
||||
for mode in ['true', 'false']:
|
||||
with patch.dict(os.environ, MINEROPS_HUB_MODE=mode):
|
||||
for path, body in [
|
||||
('/api/tools/legacy-summary-refresh-one', {'miner_id': str(a), 'ip_address': '198.51.100.122'}),
|
||||
('/api/tools/legacy-summary-refresh-one', {'miner_id': str(a)}),
|
||||
('/api/tools/legacy-summary-refresh-missing', {}),
|
||||
]:
|
||||
before = self.durable()
|
||||
self.denied(self.client.post(path, headers=self.admin, json=body), before, 410)
|
||||
read.assert_not_called()
|
||||
|
||||
def test_summary_cannot_start_inflight_read_or_overwrite_newer_owner(self):
|
||||
mid, _ = self.miner(ip='198.51.100.123')
|
||||
with patch('app.main.read_legacy_whatsminer_summary') as read, ThreadPoolExecutor(2) as pool:
|
||||
future = pool.submit(lambda: self.client.post('/api/tools/legacy-summary-refresh-one', headers=self.admin, json={'miner_id': str(mid)}))
|
||||
with connect() as c:
|
||||
c.execute("UPDATE miners SET site_id=%s,agent_zone_id=%s,current_ip='198.51.100.124',updated_at=clock_timestamp() WHERE id=%s", (self.sites[1], self.zones[1], mid))
|
||||
self.assertEqual(future.result(timeout=8).status_code, 410)
|
||||
before = self.durable()
|
||||
with self.assertRaises(HTTPException):
|
||||
main.apply_legacy_summary_to_miner(str(mid), '198.51.100.123')
|
||||
self.assertEqual(before, self.durable())
|
||||
self.assertEqual(self.scalar('SELECT host(current_ip) FROM miners WHERE id=%s', (mid,)), '198.51.100.124')
|
||||
read.assert_not_called()
|
||||
|
||||
def test_legacy_human_auth_still_required(self):
|
||||
for headers in [{}, self.headers()]:
|
||||
for path in ['/api/miners/discover-one', '/api/tools/legacy-summary-refresh-one']:
|
||||
before = self.durable()
|
||||
self.denied(self.client.post(path, headers=headers, json={'ip_address': '198.51.100.111'}), before, 401)
|
||||
|
||||
def test_csv_ip_only_requires_zone_even_with_one_global_match(self):
|
||||
self.miner(1, ip='198.51.100.112')
|
||||
before = self.durable()
|
||||
self.denied(self.upload('IP,Miner Type\n198.51.100.112,WRONG\n'), before, 422)
|
||||
|
||||
def test_csv_same_ip_requires_namespace_and_exact_zone_updates_one(self):
|
||||
a, _ = self.miner(ip='198.51.100.126')
|
||||
b, _ = self.miner(1, ip='198.51.100.126')
|
||||
before = self.durable()
|
||||
self.denied(self.upload('IP,Miner Type\n198.51.100.126,WRONG\n'), before, 422)
|
||||
r = self.upload(f'IP,Agent Zone ID,Miner Type\n198.51.100.126,{self.zones[1]},VALID\n')
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertIsNone(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (a,)))
|
||||
self.assertEqual(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (b,)), 'VALID')
|
||||
self.assertEqual(before['miner_status_snapshots'], self.durable()['miner_status_snapshots'])
|
||||
|
||||
def test_csv_duplicate_ip_inside_zone_rejected(self):
|
||||
self.miner(ip='198.51.100.130'); self.miner(ip='198.51.100.130')
|
||||
before = self.durable()
|
||||
self.denied(self.upload('IP,Miner Type\n198.51.100.130,WRONG\n', '&agent_zone_id=' + str(self.zones[0])), before, 409)
|
||||
|
||||
def test_csv_exact_id_mac_serial_and_query_zone_legitimate(self):
|
||||
mid, mac = self.miner(ip='198.51.100.131')
|
||||
serial = uuid4().hex
|
||||
with connect() as c:
|
||||
c.execute('UPDATE miners SET miner_sn=%s WHERE id=%s', (serial, mid))
|
||||
for heading, value in [('Miner ID', str(mid)), ('MAC', mac), ('Serial', serial), ('IP', '198.51.100.131')]:
|
||||
r = self.upload(f'{heading},Miner Type\n{value},VALID\n', '&agent_zone_id=' + str(self.zones[0]))
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertEqual(r.json()['data']['matched_count'], 1)
|
||||
|
||||
def test_csv_conflicting_and_unknown_mac_never_fall_back_to_ip(self):
|
||||
a, _ = self.miner(ip='198.51.100.132')
|
||||
b, bm = self.miner(1, ip='198.51.100.133')
|
||||
for mac in [bm, '02:AA:BB:CC:DD:EF']:
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Miner ID,IP,MAC,Miner Type\n{a},198.51.100.132,{mac},WRONG\n'), before, 409)
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Miner ID,IP,Miner Type\n{a},198.51.100.133,WRONG\n'), before, 409)
|
||||
|
||||
def test_csv_historical_mac_and_contradictory_namespace_serial_denied(self):
|
||||
mid, oldmac = self.miner(ip='198.51.100.134')
|
||||
with connect() as c:
|
||||
c.execute('UPDATE miners SET current_mac=%s,miner_sn=%s WHERE id=%s', ('02:' + ':'.join(f'{v:02X}' for v in uuid4().bytes[:5]), 'serial-' + str(mid), mid))
|
||||
for columns, values in [
|
||||
('MAC', oldmac), ('Serial', 'unknown'), ('Site ID', str(self.sites[1])),
|
||||
('Agent Zone ID', str(self.zones[1])), ('miner_id', str(uuid4())),
|
||||
]:
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Miner ID,{columns},Miner Type\n{mid},{values},WRONG\n'), before, 409)
|
||||
|
||||
def test_csv_invalid_identifiers_and_query_row_disagreement(self):
|
||||
mid, _ = self.miner(ip='198.51.100.135')
|
||||
for header, value in [('MAC', 'bad-mac'), ('IP', '198.51.100.135/24'), ('Miner ID', 'bad-id'), ('Agent Zone ID', 'bad-zone')]:
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'{header},Miner Type\n{value},WRONG\n'), before, 422)
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Miner ID,Agent Zone ID,Miner Type\n{mid},{self.zones[1]},WRONG\n', '&agent_zone_id=' + str(self.zones[0])), before, 409)
|
||||
|
||||
def test_csv_unmatched_strong_identity_does_not_select_ip(self):
|
||||
self.miner(ip='198.51.100.136')
|
||||
before = self.durable()
|
||||
r = self.upload('MAC,IP,Miner Type\n02:AA:BB:CC:DD:EF,198.51.100.136,WRONG\n')
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertEqual(r.json()['data']['unmatched_count'], 1)
|
||||
self.assertEqual(before['miners'], self.durable()['miners'])
|
||||
|
||||
def test_csv_batch_conflict_rolls_back_prior_valid_row(self):
|
||||
a, _ = self.miner(ip='198.51.100.137')
|
||||
b, _ = self.miner(1, ip='198.51.100.138')
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Miner ID,IP,Miner Type\n{a},198.51.100.137,VALID\n{b},198.51.100.137,WRONG\n'), before, 409)
|
||||
|
||||
def test_csv_preview_repeat_and_history_preservation(self):
|
||||
mid, _ = self.miner(ip='198.51.100.139')
|
||||
with connect() as c:
|
||||
c.execute('INSERT INTO miner_status_snapshots(id,miner_id,observed_at,mhs_av) VALUES (%s,%s,now(),123)', (uuid4(), mid))
|
||||
before = self.durable()
|
||||
text = f'Miner ID,Miner Type\n{mid},VALID\n'
|
||||
r = self.client.post('/api/imports/whatsminer-tool-scan?dry_run=true&update_existing=true', headers=self.admin, files={'file': ('simulated.csv', text.encode(), 'text/csv')})
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertEqual(before, self.durable())
|
||||
for _ in range(2):
|
||||
self.assertEqual(self.upload(text).status_code, 200)
|
||||
self.assertEqual(before['miner_status_snapshots'], self.durable()['miner_status_snapshots'])
|
||||
self.assertEqual(before['miner_mac_ownerships'], self.durable()['miner_mac_ownerships'])
|
||||
|
||||
def test_csv_actual_audit_permission_failure_rolls_back(self):
|
||||
mid, _ = self.miner(ip='198.51.100.140')
|
||||
role = 'b1_csv_' + uuid4().hex
|
||||
with connect() as c:
|
||||
c.execute('CREATE ROLE ' + role)
|
||||
c.execute('GRANT USAGE ON SCHEMA public TO ' + role)
|
||||
c.execute('GRANT SELECT,INSERT,UPDATE,DELETE ON ALL TABLES IN SCHEMA public TO ' + role)
|
||||
c.execute('REVOKE INSERT ON audit_logs FROM ' + role)
|
||||
def limited():
|
||||
c = connect(); c.execute('SET ROLE ' + role); c.commit(); return c
|
||||
before = self.durable()
|
||||
try:
|
||||
with patch('app.main.get_db_connection', side_effect=limited):
|
||||
r = self.upload(f'Miner ID,Miner Type\n{mid},ROLLBACK\n')
|
||||
self.denied(r, before, 500)
|
||||
self.assertIn('audit_logs', r.text)
|
||||
finally:
|
||||
with connect() as c:
|
||||
c.execute('DROP OWNED BY ' + role); c.execute('DROP ROLE ' + role)
|
||||
|
||||
def test_csv_concurrent_ownership_move_is_revalidated_after_real_lock_wait(self):
|
||||
mid, _ = self.miner(ip='198.51.100.141')
|
||||
owner = connect()
|
||||
owner.execute('UPDATE miners SET site_id=%s,agent_zone_id=%s,updated_at=clock_timestamp() WHERE id=%s', (self.sites[1], self.zones[1], mid))
|
||||
with ThreadPoolExecutor(1) as pool:
|
||||
future = pool.submit(lambda: self.upload(f'Miner ID,Agent Zone ID,Miner Type\n{mid},{self.zones[0]},WRONG\n'))
|
||||
try:
|
||||
deadline = time.monotonic() + 4
|
||||
waiting = 0
|
||||
while time.monotonic() < deadline:
|
||||
waiting = self.scalar("SELECT count(*) FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock'")
|
||||
if waiting: break
|
||||
time.sleep(.02)
|
||||
self.assertGreater(waiting, 0)
|
||||
owner.commit()
|
||||
r = future.result(timeout=8)
|
||||
self.assertEqual(r.status_code, 409, r.text)
|
||||
self.assertEqual(self.scalar('SELECT agent_zone_id FROM miners WHERE id=%s', (mid,)), self.zones[1])
|
||||
self.assertIsNone(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (mid,)))
|
||||
finally:
|
||||
owner.rollback(); owner.close()
|
||||
|
||||
def test_csv_real_update_lock_prevents_mid_transaction_ownership_change(self):
|
||||
import app.discovery.csv_import as module
|
||||
mid, _ = self.miner(ip='198.51.100.142')
|
||||
entered, release = threading.Event(), threading.Event()
|
||||
original = module.resolve_inventory_row
|
||||
def gate(*args, **kwargs):
|
||||
row = original(*args, **kwargs); entered.set()
|
||||
self.assertTrue(release.wait(4)); return row
|
||||
def move():
|
||||
with connect() as c:
|
||||
c.execute('UPDATE miners SET site_id=%s,agent_zone_id=%s,updated_at=clock_timestamp() WHERE id=%s', (self.sites[1], self.zones[1], mid))
|
||||
with patch.object(module, 'resolve_inventory_row', side_effect=gate), ThreadPoolExecutor(2) as pool:
|
||||
upload = pool.submit(lambda: self.upload(f'Miner ID,Agent Zone ID,Miner Type\n{mid},{self.zones[0]},VALID\n'))
|
||||
self.assertTrue(entered.wait(4)); moving = pool.submit(move)
|
||||
try:
|
||||
deadline = time.monotonic() + 3
|
||||
waiting = 0
|
||||
while time.monotonic() < deadline:
|
||||
waiting = self.scalar("SELECT count(*) FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock'")
|
||||
if waiting: break
|
||||
time.sleep(.02)
|
||||
self.assertGreater(waiting, 0)
|
||||
finally:
|
||||
release.set()
|
||||
self.assertEqual(upload.result(timeout=8).status_code, 200)
|
||||
moving.result(timeout=8)
|
||||
self.assertEqual(self.scalar('SELECT agent_zone_id FROM miners WHERE id=%s', (mid,)), self.zones[1])
|
||||
|
||||
def test_csv_duplicate_headers_and_conflicting_aliases_rejected(self):
|
||||
mid, _ = self.miner(ip='198.51.100.143')
|
||||
for text, code in [
|
||||
(f'Miner ID,Miner ID,Miner Type\n{mid},{mid},WRONG\n', 422),
|
||||
(f'Miner ID,miner_id,Miner Type\n{mid},{uuid4()},WRONG\n', 409),
|
||||
(f'Miner ID,Site ID,Agent Zone ID,Miner Type\n{mid},{self.sites[1]},{self.zones[0]},WRONG\n', 409),
|
||||
]:
|
||||
before = self.durable()
|
||||
self.denied(self.upload(text), before, code)
|
||||
|
||||
def test_csv_duplicate_serial_is_ambiguous(self):
|
||||
a, _ = self.miner(ip='198.51.100.144')
|
||||
b, _ = self.miner(1, ip='198.51.100.145')
|
||||
serial = 'duplicate-' + uuid4().hex
|
||||
with connect() as c:
|
||||
c.execute('UPDATE miners SET miner_sn=%s WHERE id IN (%s,%s)', (serial, a, b))
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Serial,Miner Type\n{serial},WRONG\n'), before, 409)
|
||||
|
||||
def test_csv_global_mac_selection_with_overlapping_ip_is_valid(self):
|
||||
a, _ = self.miner(ip='198.51.100.146')
|
||||
b, mac = self.miner(1, ip='198.51.100.146')
|
||||
before = self.durable()
|
||||
r = self.upload(f'MAC,IP,Miner Type\n{mac},198.51.100.146,VALID\n')
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertIsNone(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (a,)))
|
||||
self.assertEqual(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (b,)), 'VALID')
|
||||
after = self.durable()
|
||||
for table in ['miner_status_snapshots', 'miner_mac_ownerships', 'miner_network_identities']:
|
||||
self.assertEqual(before[table], after[table])
|
||||
audit = self.scalar("SELECT new_value_json FROM audit_logs WHERE actor_user_id=%s AND action='import.whatsminertool_csv.applied' ORDER BY created_at DESC LIMIT 1", (self.uid,))
|
||||
self.assertEqual(audit['updates'][0]['miner_id'], str(b))
|
||||
self.assertEqual(audit['updates'][0]['agent_zone_id'], str(self.zones[1]))
|
||||
|
||||
def test_csv_read_only_user_cannot_apply_metadata(self):
|
||||
uid = uuid4()
|
||||
with connect() as c:
|
||||
c.execute("INSERT INTO users(id,username,display_name,password_hash,role_id) SELECT %s,%s,'Read only','unused',id FROM roles WHERE name='Read Only'", (uid, str(uid)))
|
||||
headers = {'Authorization': 'Bearer ' + create_access_token(str(uid), str(uid), 'Read Only')}
|
||||
mid, _ = self.miner(ip='198.51.100.147')
|
||||
before = self.durable()
|
||||
self.denied(self.upload(f'Miner ID,Miner Type\n{mid},WRONG\n', headers=headers), before, 403)
|
||||
@@ -71,64 +71,47 @@ class FakeRepository:
|
||||
|
||||
|
||||
class MinerControlServiceTests(unittest.TestCase):
|
||||
def test_supported_projection_is_factual_and_bounded(self):
|
||||
def test_disabled_projection_preserves_facts_and_bounded_history(self):
|
||||
repository = FakeRepository()
|
||||
result = MinerControlService(repository).get_controls(MINER_ID)
|
||||
self.assertEqual(result.service_control.state, "supported")
|
||||
self.assertEqual(result.service_control.state, "unavailable")
|
||||
self.assertEqual(result.service_control.reason, "physical_control_disabled_stage_a")
|
||||
self.assertEqual(result.performance_mode, "normal")
|
||||
self.assertEqual(result.available_service_action, "suspend")
|
||||
self.assertIsNone(result.available_service_action)
|
||||
self.assertEqual(repository.limits, [5])
|
||||
|
||||
def test_capability_missing_facts_is_not_optimistic(self):
|
||||
cases = (
|
||||
({"agent_zone_id": None}, "unavailable", "agent_zone_unassigned"),
|
||||
({"agent_zone_exists": False}, "unavailable", "agent_zone_missing"),
|
||||
({"agent_zone_enabled": False}, "unavailable", "agent_zone_disabled"),
|
||||
({"agent_id": None}, "unavailable", "agent_unavailable"),
|
||||
({"current_ip": None}, "unavailable", "current_ip_missing"),
|
||||
({"api3_supported": False}, "unavailable", "api3_unsupported"),
|
||||
({"api3_supported": None}, "unknown", "api3_support_unknown"),
|
||||
({"last_payload": {}}, "unknown", "agent_capability_unknown"),
|
||||
(
|
||||
{"last_payload": {"control_available": False,
|
||||
"control_unavailable_reason": "credentials_missing"}},
|
||||
"unavailable", "credentials_missing",
|
||||
),
|
||||
)
|
||||
for changes, state, reason in cases:
|
||||
with self.subTest(reason=reason):
|
||||
result = MinerControlService(FakeRepository(context(**changes))).get_controls(MINER_ID)
|
||||
self.assertEqual(result.service_control.state, state)
|
||||
self.assertEqual(result.service_control.reason, reason)
|
||||
def test_capability_cannot_be_enabled_by_agent_claims(self):
|
||||
for facts in ({}, {"control_available": True}, {"control_available": False}):
|
||||
repository = FakeRepository(context(last_payload=facts))
|
||||
result = MinerControlService(repository).get_controls(MINER_ID)
|
||||
self.assertEqual(result.service_control.state, "unavailable")
|
||||
self.assertEqual(result.service_control.reason, "physical_control_disabled_stage_a")
|
||||
|
||||
def test_resume_requires_known_stopped_state(self):
|
||||
service = MinerControlService(FakeRepository(context(runtime_state="offline")))
|
||||
with self.assertRaisesRegex(MinerControlUnavailable, "factually_stopped"):
|
||||
service.queue_control(
|
||||
MINER_ID, "resume", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="operator",
|
||||
)
|
||||
def test_resume_is_disabled_even_for_known_stopped_state(self):
|
||||
service = MinerControlService(FakeRepository(context(runtime_state="stopped")))
|
||||
with self.assertRaisesRegex(MinerControlUnavailable, "physical_control_disabled_stage_a"):
|
||||
service.queue_control(MINER_ID, "resume", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="operator")
|
||||
|
||||
def test_queue_contains_routing_but_audit_and_response_do_not(self):
|
||||
def test_all_current_actions_are_rejected_before_routing_or_queue_write(self):
|
||||
for action in ("suspend", "resume", "performance_mode"):
|
||||
repository = FakeRepository()
|
||||
with self.assertRaisesRegex(MinerControlUnavailable, "physical_control_disabled_stage_a"):
|
||||
MinerControlService(repository).queue_control(
|
||||
MINER_ID, action, requested_value="high",
|
||||
actor_user_id=USER_ID, actor_username="operator")
|
||||
self.assertEqual(repository.enqueue_calls, [])
|
||||
self.assertFalse(hasattr(repository, "requested_miner"))
|
||||
|
||||
def test_rejected_request_leaves_context_unchanged(self):
|
||||
repository = FakeRepository()
|
||||
result = MinerControlService(repository).queue_control(
|
||||
MINER_ID, "performance_mode", requested_value="high",
|
||||
actor_user_id=USER_ID, actor_username="operator",
|
||||
)
|
||||
call = repository.enqueue_calls[0]
|
||||
self.assertEqual(call["request_payload"]["ip_address"], "192.0.2.10")
|
||||
self.assertNotIn("ip_address", call["audit_payload"])
|
||||
self.assertNotIn("agent_id", call["audit_payload"])
|
||||
self.assertNotIn("password", str(call).lower())
|
||||
self.assertNotIn("192.0.2.10", result.model_dump_json())
|
||||
|
||||
def test_no_projection_changes_when_work_is_queued(self):
|
||||
repository = FakeRepository()
|
||||
MinerControlService(repository).queue_control(
|
||||
MINER_ID, "suspend", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="operator",
|
||||
)
|
||||
self.assertFalse(any("update" in key for key in repository.enqueue_calls[0]))
|
||||
before = repository.miner_context.copy()
|
||||
with self.assertRaises(MinerControlUnavailable):
|
||||
MinerControlService(repository).queue_control(
|
||||
MINER_ID, "suspend", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="operator")
|
||||
self.assertEqual(repository.miner_context, before)
|
||||
self.assertEqual(repository.enqueue_calls, [])
|
||||
|
||||
|
||||
class MinerControlRouterTests(unittest.TestCase):
|
||||
@@ -159,18 +142,16 @@ class MinerControlRouterTests(unittest.TestCase):
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.json()["data"]["miner_id"], str(MINER_ID))
|
||||
|
||||
def test_posts_require_control_and_return_202_bounded_envelope(self):
|
||||
path = f"/api/miners/{MINER_ID}/controls/suspend"
|
||||
self.authenticate(("miners.view",))
|
||||
self.assertEqual(self.client.post(path, json={}).status_code, 403)
|
||||
self.authenticate(("miners.control",))
|
||||
response = self.client.post(path, json={})
|
||||
self.assertEqual(response.status_code, 202)
|
||||
self.assertEqual(set(response.json()["data"]), {
|
||||
"operation_id", "miner_id", "action", "requested_value", "status", "created_at"
|
||||
})
|
||||
self.assertNotIn("ip", response.text.lower())
|
||||
self.assertNotIn("agent", response.text.lower())
|
||||
def test_posts_require_control_and_then_fail_closed(self):
|
||||
for action, body in (("suspend", {}), ("resume", {}), ("performance-mode", {"mode": "high"})):
|
||||
path = f"/api/miners/{MINER_ID}/controls/{action}"
|
||||
self.authenticate(("miners.view",))
|
||||
self.assertEqual(self.client.post(path, json=body).status_code, 403)
|
||||
self.authenticate(("miners.control",))
|
||||
response = self.client.post(path, json=body)
|
||||
self.assertEqual(response.status_code, 409)
|
||||
self.assertIn("physical_control_disabled_stage_a", response.text)
|
||||
self.assertEqual(self.repository.enqueue_calls, [])
|
||||
|
||||
def test_strict_bodies_reject_routing_commands_and_credentials(self):
|
||||
self.authenticate(("miners.control",))
|
||||
@@ -195,7 +176,7 @@ class MinerControlRouterTests(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(
|
||||
self.client.post(path, json={}, headers={"Origin": "http://testserver"}).status_code,
|
||||
202,
|
||||
409,
|
||||
)
|
||||
|
||||
def test_identity_permission_mapping_is_specific(self):
|
||||
|
||||
@@ -137,7 +137,7 @@ class MinerControlPostgresTests(unittest.TestCase):
|
||||
"""
|
||||
INSERT INTO minerops_agent_command_queue (
|
||||
id, agent_id, command_type, command_status, request_payload, created_at
|
||||
) VALUES (%s, %s, 'miner_control', %s, %s, COALESCE(%s, now()))
|
||||
) VALUES (%s, %s, 'poll_known_miners_now', %s, %s, COALESCE(%s, now()))
|
||||
""",
|
||||
(command_id, AGENT_ID, status,
|
||||
Jsonb({"schema_version": 1, "miner_id": str(miner_id),
|
||||
@@ -199,13 +199,22 @@ class MinerControlPostgresTests(unittest.TestCase):
|
||||
service.submit_result(AGENT_ID, pending)
|
||||
|
||||
def test_atomic_queue_audit_duplicate_live_and_completed_history(self):
|
||||
service = MinerControlService(MinerControlRepository(connect))
|
||||
first = service.queue_control(
|
||||
repository = MinerControlRepository(connect)
|
||||
def queue_control(miner_id, action, *, requested_value, actor_user_id, actor_username):
|
||||
from types import SimpleNamespace
|
||||
from uuid import uuid4
|
||||
return SimpleNamespace(**repository.enqueue_with_audit(
|
||||
operation_id=str(uuid4()), miner_id=miner_id, agent_id=AGENT_ID,
|
||||
actor_user_id=actor_user_id, actor_username=actor_username,
|
||||
request_payload={"schema_version": 1, "miner_id": str(miner_id),
|
||||
"action": action, "requested_value": requested_value},
|
||||
audit_payload={"miner_id": str(miner_id), "action": action}))
|
||||
first = queue_control(
|
||||
MINER_ID, "suspend", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="control-test-user",
|
||||
)
|
||||
with self.assertRaises(MinerControlConflict):
|
||||
service.queue_control(
|
||||
queue_control(
|
||||
MINER_ID, "suspend", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="control-test-user",
|
||||
)
|
||||
@@ -223,17 +232,26 @@ class MinerControlPostgresTests(unittest.TestCase):
|
||||
connection.commit()
|
||||
finally:
|
||||
connection.close()
|
||||
second = service.queue_control(
|
||||
second = queue_control(
|
||||
MINER_ID, "suspend", requested_value=None,
|
||||
actor_user_id=USER_ID, actor_username="control-test-user",
|
||||
)
|
||||
self.assertNotEqual(first.operation_id, second.operation_id)
|
||||
|
||||
def test_audit_failure_rolls_back_queue_row(self):
|
||||
service = MinerControlService(MinerControlRepository(connect))
|
||||
repository = MinerControlRepository(connect)
|
||||
def queue_control(miner_id, action, *, requested_value, actor_user_id, actor_username):
|
||||
from types import SimpleNamespace
|
||||
from uuid import uuid4
|
||||
return SimpleNamespace(**repository.enqueue_with_audit(
|
||||
operation_id=str(uuid4()), miner_id=miner_id, agent_id=AGENT_ID,
|
||||
actor_user_id=actor_user_id, actor_username=actor_username,
|
||||
request_payload={"schema_version": 1, "miner_id": str(miner_id),
|
||||
"action": action, "requested_value": requested_value},
|
||||
audit_payload={"miner_id": str(miner_id), "action": action}))
|
||||
operation_count_before = self._operation_count()
|
||||
with self.assertRaises(Exception):
|
||||
service.queue_control(
|
||||
queue_control(
|
||||
MINER_ID, "suspend", requested_value=None,
|
||||
actor_user_id=UUID("00000000-0000-4000-8000-000000009999"),
|
||||
actor_username="missing-user",
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
"""Defense-in-depth unit tests; no physical miner or external service is used."""
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import unittest
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app import main
|
||||
from app.agent_commands.exceptions import AgentCommandConflict
|
||||
from app.agent_commands.models import AgentCommandResultRequest
|
||||
from app.agent_commands.policy import rejection_reason
|
||||
from app.agent_commands.service import AgentCommandService
|
||||
|
||||
|
||||
class StageAContainmentTests(unittest.TestCase):
|
||||
def test_service_dispatch_rejects_physical_and_unknown_rows_even_if_repository_regresses(self):
|
||||
repository = Mock()
|
||||
repository.claim_pending.return_value = [
|
||||
{"command_id": str(index), "command_type": kind, "request_payload": payload,
|
||||
"created_at": datetime.now(timezone.utc)}
|
||||
for index, (kind, payload) in enumerate((
|
||||
("miner_control", {"schema_version": 1, "action": "suspend"}),
|
||||
("miner_control", {"schema_version": 999}),
|
||||
("miner_control_v2", {}), ("raw_api3", {}),
|
||||
("poll_known_miners_now", {}),
|
||||
))
|
||||
]
|
||||
result = AgentCommandService(repository).claim("simulated-agent")
|
||||
self.assertEqual([row.command_type for row in result], ["poll_known_miners_now"])
|
||||
|
||||
def test_service_result_rejection_does_not_depend_on_payload_validation(self):
|
||||
for kind in ("miner_control", "raw_api3", "miner_control_v2"):
|
||||
repository = Mock()
|
||||
repository.get_command.return_value = {"id": "old-command", "agent_id": "simulated-agent",
|
||||
"command_type": kind, "command_status": "claimed", "request_payload": {}}
|
||||
with self.assertRaises(AgentCommandConflict):
|
||||
AgentCommandService(repository).submit_result("simulated-agent",
|
||||
AgentCommandResultRequest(command_id="old-command", status="completed",
|
||||
result_payload={"schema_version": 999, "command_accepted": True}))
|
||||
args = repository.finish_command.call_args.args
|
||||
self.assertEqual(args[2:4], ("cancelled", {}))
|
||||
self.assertEqual(args[4], rejection_reason(kind))
|
||||
|
||||
def test_dormant_legacy_helpers_cannot_transmit_or_queue_if_called_internally(self):
|
||||
names = (
|
||||
"dry_run_reboot_miner_command", "change_miner_api3_password",
|
||||
"minerops_queue_agent_miner_control", "minerops_power_command",
|
||||
"minerops_send_whatsminer_reboot_sync", "minerops_send_whatsminer_reboot",
|
||||
"minerops_send_whatsminer_command_sync", "minerops_api3_send_payload_sync",
|
||||
"minerops_api3_service_sync", "minerops_api3_service",
|
||||
"minerops_send_control_command_item", "minerops_send_whatsminer_command_sequence",
|
||||
"minerops_send_whatsminer_command", "minerops_api3_service_sync_with_credentials",
|
||||
"minerops_api3_service_with_credentials", "minerops_api3_change_password_sync",
|
||||
"minerops_api3_change_password",
|
||||
)
|
||||
with patch.object(main.socket, "create_connection") as transport, patch.object(main, "get_db_connection") as database, patch.object(main, "minerops_command_db_connect") as async_database:
|
||||
for name in names:
|
||||
function = getattr(main, name)
|
||||
required = ["simulated" for parameter in inspect.signature(function).parameters.values()
|
||||
if parameter.default is inspect.Parameter.empty]
|
||||
with self.subTest(name=name), self.assertRaises(HTTPException) as rejected:
|
||||
result = function(*required)
|
||||
if inspect.isawaitable(result):
|
||||
asyncio.run(result)
|
||||
self.assertEqual(rejected.exception.status_code, 410)
|
||||
transport.assert_not_called()
|
||||
database.assert_not_called()
|
||||
async_database.assert_not_called()
|
||||
|
||||
def test_fixed_internal_discovery_queries_still_use_only_allowlisted_selection(self):
|
||||
parameters = {"get_miner_info": {"info": "ip,proto,netmask,gateway,dns,hostname,mac,ledstat,minersn,powersn"},
|
||||
"get_version": {}, "summary": {}, "edevs": {}, "pools": {}}
|
||||
for command, params in parameters.items():
|
||||
with patch.object(main.socket, "create_connection") as transport:
|
||||
socket = transport.return_value.__enter__.return_value
|
||||
socket.recv.side_effect = [b'{"STATUS":[]}', b'']
|
||||
result = main.send_whatsminer_read_command("192.0.2.1", 4028, command, 500, params)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(socket.sendall.call_count, 1)
|
||||
import json
|
||||
transmitted = json.loads(socket.sendall.call_args.args[0])
|
||||
self.assertEqual(transmitted, {"command": command, **params})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,206 @@
|
||||
"""Stage A queue containment; requires an explicitly disposable PostgreSQL DB."""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from unittest.mock import patch
|
||||
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
from app.agent_commands.exceptions import AgentCommandConflict
|
||||
from app.agent_commands.models import AgentCommandResultRequest
|
||||
from app.agent_commands.repository import AgentCommandRepository
|
||||
from app.agent_commands.service import AgentCommandService
|
||||
from app.config import settings
|
||||
|
||||
|
||||
AGENT = "stage-a-simulated-agent"
|
||||
OTHER_AGENT = "stage-a-other-agent"
|
||||
|
||||
|
||||
def connect():
|
||||
return psycopg.connect(host=settings.postgres_host, port=settings.postgres_port,
|
||||
dbname=settings.postgres_db, user=settings.postgres_user,
|
||||
password=settings.postgres_password, connect_timeout=5)
|
||||
|
||||
|
||||
class StageAQueueContainmentTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if os.environ.get("MINEROPS_STAGE_A_DISPOSABLE") != "1" or not settings.postgres_db.startswith("minerops_stage_a_"):
|
||||
raise RuntimeError("Refusing queue tests outside an explicitly disposable Stage A database")
|
||||
|
||||
def setUp(self):
|
||||
self.transport_patch = patch("socket.create_connection", side_effect=AssertionError("No miner transport allowed"))
|
||||
self.transport = self.transport_patch.start()
|
||||
self.addCleanup(self.transport_patch.stop)
|
||||
self.addCleanup(self.transport.assert_not_called)
|
||||
with connect() as connection, connection.cursor() as cur:
|
||||
cur.execute("DELETE FROM audit_logs WHERE entity_type = 'agent_command' AND old_value_json ->> 'agent_id' IN (%s, %s)", (AGENT, OTHER_AGENT))
|
||||
cur.execute("DELETE FROM minerops_agent_command_queue WHERE agent_id IN (%s, %s)", (AGENT, OTHER_AGENT))
|
||||
self.repository = AgentCommandRepository(connect)
|
||||
self.service = AgentCommandService(self.repository)
|
||||
|
||||
@staticmethod
|
||||
def seed(command_id, status="pending", command_type="miner_control", payload=None, agent=AGENT):
|
||||
if payload is None:
|
||||
payload = {"action": "legacy_4028", "ip_address": "192.0.2.1", "legacy_4028_command": "reboot", "dry_run": False}
|
||||
with connect() as connection, connection.cursor() as cur:
|
||||
cur.execute("""INSERT INTO minerops_agent_command_queue
|
||||
(id, agent_id, command_type, command_status, request_payload,
|
||||
result_payload, error_message, claimed_at)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, 'prior evidence',
|
||||
CASE WHEN %s = 'claimed' THEN now() ELSE NULL END)""",
|
||||
(command_id, agent, command_type, status, Jsonb(payload),
|
||||
Jsonb({"historical_evidence": True}), status))
|
||||
|
||||
@staticmethod
|
||||
def row(command_id):
|
||||
with connect() as connection, connection.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("SELECT * FROM minerops_agent_command_queue WHERE id = %s", (command_id,))
|
||||
return dict(cur.fetchone())
|
||||
|
||||
@staticmethod
|
||||
def audits(command_id):
|
||||
with connect() as connection, connection.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("SELECT * FROM audit_logs WHERE action = 'agent.command.rejected' AND old_value_json ->> 'command_id' = %s", (command_id,))
|
||||
return [dict(row) for row in cur.fetchall()]
|
||||
|
||||
def assert_cancelled_preserving_history(self, command_id, before, outcome):
|
||||
after = self.row(command_id)
|
||||
self.assertEqual(after["command_status"], "cancelled")
|
||||
for field in ("id", "agent_id", "command_type", "request_payload", "result_payload", "created_at", "claimed_at"):
|
||||
self.assertEqual(after[field], before[field], field)
|
||||
self.assertTrue(after["error_message"].startswith("prior evidence\n"))
|
||||
self.assertIsNotNone(after["completed_at"])
|
||||
audits = self.audits(command_id)
|
||||
self.assertEqual(len(audits), 1)
|
||||
self.assertEqual(audits[0]["old_value_json"]["status"], before["command_status"])
|
||||
self.assertEqual(audits[0]["new_value_json"]["execution_outcome"], outcome)
|
||||
self.assertNotIn("192.0.2.1", str(audits))
|
||||
|
||||
def test_seeded_legacy_pending_never_dispatched(self):
|
||||
self.seed("stage-a-pending")
|
||||
before = self.row("stage-a-pending")
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
self.assert_cancelled_preserving_history("stage-a-pending", before, "unknown")
|
||||
|
||||
def test_seeded_legacy_claimed_cancelled_with_unknown_prior_execution(self):
|
||||
self.seed("stage-a-claimed", status="claimed")
|
||||
before = self.row("stage-a-claimed")
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
self.assert_cancelled_preserving_history("stage-a-claimed", before, "unknown")
|
||||
|
||||
def test_all_physical_schemas_and_unknown_types_are_rejected(self):
|
||||
payloads = ({}, {"schema_version": 1, "action": "suspend"},
|
||||
{"schema_version": 2, "action": "resume"},
|
||||
{"schema_version": "1", "action": "performance_mode"},
|
||||
{"schema_version": 999, "action": "reboot"})
|
||||
for index, payload in enumerate(payloads):
|
||||
self.seed(f"stage-a-schema-{index}", payload=payload)
|
||||
self.seed("stage-a-unknown", command_type="raw_api3")
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
for index in range(len(payloads)):
|
||||
self.assertIn("physical_control_disabled_stage_a", self.row(f"stage-a-schema-{index}")["error_message"])
|
||||
self.assertIn("unsupported_command_schema_stage_a", self.row("stage-a-unknown")["error_message"])
|
||||
|
||||
def test_late_results_and_replay_cannot_promote_old_claimed_commands(self):
|
||||
self.seed("stage-a-late", status="claimed")
|
||||
before = self.row("stage-a-late")
|
||||
for status in ("completed", "failed", "completed"):
|
||||
request = AgentCommandResultRequest(command_id="stage-a-late", status=status,
|
||||
result_payload={"command_accepted": True})
|
||||
with self.assertRaisesRegex(AgentCommandConflict, "physical_control_disabled_stage_a"):
|
||||
self.service.submit_result(AGENT, request)
|
||||
self.assert_cancelled_preserving_history("stage-a-late", before, "unknown")
|
||||
|
||||
def test_repository_cannot_be_bypassed_by_direct_finish_call(self):
|
||||
self.seed("stage-a-direct", status="pending")
|
||||
before = self.row("stage-a-direct")
|
||||
with self.assertRaises(AgentCommandConflict):
|
||||
self.repository.finish_command("stage-a-direct", AGENT, "completed", {"accepted": True}, None)
|
||||
self.assert_cancelled_preserving_history("stage-a-direct", before, "unknown")
|
||||
|
||||
def test_wrong_agent_cannot_cancel_or_promote_another_agents_entry(self):
|
||||
self.seed("stage-a-foreign", status="claimed", agent=OTHER_AGENT)
|
||||
before = self.row("stage-a-foreign")
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
with self.assertRaises(AgentCommandConflict):
|
||||
self.service.submit_result(AGENT, AgentCommandResultRequest(command_id="stage-a-foreign", status="completed"))
|
||||
with self.assertRaises(AgentCommandConflict):
|
||||
self.repository.finish_command("stage-a-foreign", AGENT, "completed", {}, None)
|
||||
self.assertEqual(self.row("stage-a-foreign"), before)
|
||||
self.assertEqual(self.audits("stage-a-foreign"), [])
|
||||
|
||||
def test_terminal_history_is_immutable_and_not_reaudited(self):
|
||||
for status in ("completed", "failed", "cancelled"):
|
||||
command_id = f"stage-a-terminal-{status}"
|
||||
self.seed(command_id, status=status)
|
||||
before = self.row(command_id)
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
with self.assertRaises(AgentCommandConflict):
|
||||
self.repository.finish_command(command_id, AGENT, "completed", {}, None)
|
||||
self.assertEqual(self.row(command_id), before)
|
||||
self.assertEqual(self.audits(command_id), [])
|
||||
|
||||
def test_read_only_commands_still_claim_and_complete(self):
|
||||
from uuid import uuid4
|
||||
from app.discovery.repository import DiscoveryRepository
|
||||
site, zone, actor = uuid4(), uuid4(), uuid4()
|
||||
with connect() as connection, connection.cursor() as cur:
|
||||
cur.execute("INSERT INTO sites(id,name) VALUES (%s,%s)",(site,str(site)))
|
||||
cur.execute("INSERT INTO agent_zones(id,site_id,name) VALUES (%s,%s,'Stage A reader')",(zone,site))
|
||||
cur.execute("INSERT INTO agent_zone_network_ranges(agent_zone_id,network) VALUES (%s,'192.0.2.0/24')",(zone,))
|
||||
cur.execute("""INSERT INTO minerops_site_agents(agent_id,site_id,agent_zone_id,site_name)
|
||||
VALUES (%s,%s,%s,'Stage A reader') ON CONFLICT(agent_id) DO UPDATE SET
|
||||
site_id=EXCLUDED.site_id,agent_zone_id=EXCLUDED.agent_zone_id""",(AGENT,site,zone))
|
||||
cur.execute("INSERT INTO users(id,username,display_name,password_hash,role_id) SELECT %s,%s,'Reader','unused',id FROM roles WHERE name='Admin'",(actor,str(actor)))
|
||||
discovery=DiscoveryRepository(connect)
|
||||
observation=discovery.ingest(AGENT,{'discoveries':[{'ip_address':'192.0.2.1'}]})['discoveries'][0]
|
||||
user={'id':str(actor),'username':'stage-a-reader'}
|
||||
for kind in ('poll_known_miners_now','discover_new_miners'):
|
||||
discovery.queue({'agent_id':AGENT},user,kind)
|
||||
discovery.queue({'discovery_id':observation['id'],'version':observation['version']},user)
|
||||
commands = self.service.claim(AGENT)
|
||||
self.assertEqual(len(commands), 3)
|
||||
for command in commands:
|
||||
self.service.submit_result(AGENT, AgentCommandResultRequest(command_id=command.command_id, status="completed", result_payload={"simulated": True}))
|
||||
self.assertEqual(self.row(command.command_id)["command_status"], "completed")
|
||||
|
||||
def test_concurrent_pollers_cancel_once_and_never_deliver_physical_commands(self):
|
||||
self.seed("stage-a-concurrent")
|
||||
before = self.row("stage-a-concurrent")
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
results = list(pool.map(lambda _: AgentCommandRepository(connect).claim_pending(AGENT), range(2)))
|
||||
self.assertEqual(results, [[], []])
|
||||
self.assert_cancelled_preserving_history("stage-a-concurrent", before, "unknown")
|
||||
|
||||
def test_locked_legacy_entry_cannot_leak_into_claim_query(self):
|
||||
self.seed("stage-a-locked")
|
||||
locker = connect()
|
||||
try:
|
||||
with locker.cursor() as cur:
|
||||
cur.execute("SELECT id FROM minerops_agent_command_queue WHERE id = 'stage-a-locked' FOR UPDATE")
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
finally:
|
||||
locker.rollback()
|
||||
locker.close()
|
||||
self.assertEqual(self.service.claim(AGENT), [])
|
||||
self.assertEqual(self.row("stage-a-locked")["command_status"], "cancelled")
|
||||
|
||||
def test_audit_failure_rolls_back_cancellation_and_prevents_dispatch(self):
|
||||
self.seed("stage-a-audit-failure")
|
||||
before = self.row("stage-a-audit-failure")
|
||||
def fail_audit(cur, row, **kwargs):
|
||||
cur.execute("SELECT 1 / 0")
|
||||
with patch.object(AgentCommandRepository, "_audit_rejection", side_effect=fail_audit):
|
||||
with self.assertRaises(psycopg.errors.DivisionByZero):
|
||||
self.service.claim(AGENT)
|
||||
self.assertEqual(self.row("stage-a-audit-failure"), before)
|
||||
self.assertEqual(self.audits("stage-a-audit-failure"), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,21 @@
|
||||
"""Conservative delivery outcome decisions, independent of SQL row shape."""
|
||||
import unittest
|
||||
from app.agent_commands.policy import rejection_execution_outcome
|
||||
|
||||
|
||||
class DeliveryOutcomeTests(unittest.TestCase):
|
||||
def test_only_explicit_clean_pending_evidence_allows_not_dispatched(self):
|
||||
self.assertEqual(rejection_execution_outcome({"previous_status": "pending", "no_delivery_evidence": True}), "not_dispatched")
|
||||
|
||||
def test_missing_ambiguous_or_reported_delivery_is_unknown(self):
|
||||
for row in ({}, {"previous_status": "pending"},
|
||||
{"previous_status": "pending", "no_delivery_evidence": False},
|
||||
{"previous_status": "pending", "no_delivery_evidence": None},
|
||||
{"previous_status": "pending", "no_delivery_evidence": 1},
|
||||
{"previous_status": "pending", "no_delivery_evidence": "true"},
|
||||
{"previous_status": "claimed", "no_delivery_evidence": True},
|
||||
{"previous_status": "cancelled", "no_delivery_evidence": True},
|
||||
{"previous_status": "pending", "no_delivery_evidence": True, "delivery_reported": True},
|
||||
{"previous_status": "pending", "no_delivery_evidence": True, "delivery_reported": None}):
|
||||
with self.subTest(row=row):
|
||||
self.assertEqual(rejection_execution_outcome(row), "unknown")
|
||||
@@ -0,0 +1,230 @@
|
||||
"""Delivery evidence across every Stage A retirement entry point; no miner IO."""
|
||||
import os
|
||||
import unittest
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from unittest.mock import patch
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg.types.json import Jsonb
|
||||
from app.agent_commands.exceptions import AgentCommandConflict
|
||||
from app.agent_commands.repository import AgentCommandRepository
|
||||
from app.config import settings
|
||||
from tests.test_stage_a_containment_postgres import connect
|
||||
|
||||
AGENT = "stage-a-final-evidence-agent"
|
||||
ACTOR = "00000000-0000-4000-8000-000000003302"
|
||||
SCHEDULE = "stage-a-final-evidence-schedule"
|
||||
PATHS = ("poll", "result", "cutover", "cancel", "delete")
|
||||
# Verbatim CLAIM_COMMANDS_SQL from committed 2d89e80. Test-only legacy delivery;
|
||||
# never use this query in the Stage A production repository.
|
||||
BASELINE_CLAIM_SQL = """
|
||||
WITH candidates AS (
|
||||
SELECT id
|
||||
FROM minerops_agent_command_queue
|
||||
WHERE agent_id = %s
|
||||
AND command_status = 'pending'
|
||||
ORDER BY created_at, id
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT %s
|
||||
)
|
||||
UPDATE minerops_agent_command_queue AS queue
|
||||
SET command_status = 'claimed', claimed_at = now()
|
||||
FROM candidates
|
||||
WHERE queue.id = candidates.id
|
||||
AND queue.command_status = 'pending'
|
||||
RETURNING queue.id AS command_id, queue.command_type,
|
||||
queue.request_payload, queue.created_at
|
||||
"""
|
||||
|
||||
|
||||
class DeliveryEvidencePostgresTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if os.environ.get("MINEROPS_STAGE_A_DISPOSABLE") != "1" or not settings.postgres_db.startswith("minerops_stage_a_"):
|
||||
raise RuntimeError("Delivery evidence tests require a disposable Stage A database")
|
||||
|
||||
def setUp(self):
|
||||
self.repo = AgentCommandRepository(connect)
|
||||
self.clean()
|
||||
self.addCleanup(self.clean)
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO users (id,username,display_name,password_hash,role_id)
|
||||
SELECT %s,'stage-a-final-evidence','Delivery Evidence','unused',id
|
||||
FROM roles WHERE name='Admin' ON CONFLICT (id) DO NOTHING""", (ACTOR,))
|
||||
self.no_transport = patch("socket.create_connection", side_effect=AssertionError("No miner IO"))
|
||||
mocked = self.no_transport.start()
|
||||
self.addCleanup(mocked.assert_not_called)
|
||||
self.addCleanup(self.no_transport.stop)
|
||||
|
||||
def clean(self):
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("DELETE FROM audit_logs WHERE actor_user_id=%s OR old_value_json->>'agent_id'=%s", (ACTOR, AGENT))
|
||||
cur.execute("DELETE FROM minerops_agent_command_results WHERE agent_id=%s", (AGENT,))
|
||||
cur.execute("DELETE FROM minerops_agent_command_queue WHERE agent_id=%s", (AGENT,))
|
||||
cur.execute("DELETE FROM minerops_scheduled_bulk_actions WHERE id=%s", (SCHEDULE,))
|
||||
|
||||
def seed(self, command_id="stage-a-final-evidence-command", kind="miner_control"):
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO minerops_scheduled_bulk_actions
|
||||
(id,agent_id,target_type,status,scheduled_for) VALUES (%s,%s,'all','running',now())
|
||||
ON CONFLICT (id) DO UPDATE SET status='running'""", (SCHEDULE,AGENT))
|
||||
cur.execute("""INSERT INTO minerops_agent_command_queue (id,agent_id,command_type,request_payload)
|
||||
VALUES (%s,%s,%s,%s)""",
|
||||
(command_id,AGENT,kind,Jsonb({"scheduled_action_id": SCHEDULE,"action":"suspend","ip_address":"127.0.0.1"})))
|
||||
return command_id
|
||||
|
||||
def row(self, cid):
|
||||
with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("SELECT * FROM minerops_agent_command_queue WHERE id=%s", (cid,))
|
||||
return dict(cur.fetchone())
|
||||
|
||||
def audits(self, cid):
|
||||
with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("""SELECT * FROM audit_logs WHERE action='agent.command.rejected'
|
||||
AND old_value_json->>'command_id'=%s""", (cid,))
|
||||
return list(cur.fetchall())
|
||||
|
||||
def retire(self, path, cid):
|
||||
if path == "poll": self.assertEqual(self.repo.claim_pending(AGENT), [])
|
||||
elif path == "result":
|
||||
with self.assertRaises(AgentCommandConflict): self.repo.finish_command(cid, AGENT, "completed", {}, None)
|
||||
elif path == "cutover": return self.repo.retire_unsupported(AGENT, actor_user_id=ACTOR)
|
||||
else: return self.repo.retire_schedule(SCHEDULE, actor_user_id=ACTOR, operation=path)
|
||||
|
||||
def assert_retired(self, cid, before, outcome):
|
||||
after = self.row(cid)
|
||||
self.assertEqual(after["command_status"], "cancelled")
|
||||
self.assertIsNotNone(after["completed_at"])
|
||||
for key in ("id","agent_id","command_type","claimed_at","created_at","request_payload","result_payload"):
|
||||
self.assertEqual(after[key], before[key], key)
|
||||
if before["error_message"]: self.assertTrue(after["error_message"].startswith(before["error_message"]+"\n"))
|
||||
audits = self.audits(cid)
|
||||
self.assertEqual(len(audits), 1)
|
||||
self.assertEqual(audits[0]["old_value_json"]["status"], before["command_status"])
|
||||
self.assertEqual(audits[0]["new_value_json"]["execution_outcome"], outcome)
|
||||
|
||||
def test_genuine_baseline_claim_then_explicit_repair_requeue_all_paths(self):
|
||||
for path in PATHS:
|
||||
with self.subTest(path=path):
|
||||
self.clean(); cid=self.seed()
|
||||
with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute(BASELINE_CLAIM_SQL,(AGENT,5)); delivered=list(cur.fetchall())
|
||||
self.assertEqual([r["command_id"] for r in delivered], [cid])
|
||||
claimed=self.row(cid)
|
||||
self.assertEqual(claimed["command_status"],"claimed"); self.assertIsNotNone(claimed["claimed_at"])
|
||||
# Explicitly simulate a legacy repair. No production requeue path
|
||||
# is asserted or added; retain the actual baseline claim evidence.
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("UPDATE minerops_agent_command_queue SET command_status='pending' WHERE id=%s",(cid,))
|
||||
before=self.row(cid); self.assertEqual(before["claimed_at"],claimed["claimed_at"])
|
||||
self.retire(path,cid); self.assert_retired(cid,before,"unknown")
|
||||
audit_before=self.audits(cid); row_before=self.row(cid)
|
||||
for retry in PATHS: self.retire(retry,cid)
|
||||
self.assertEqual(self.audits(cid),audit_before); self.assertEqual(self.row(cid),row_before)
|
||||
|
||||
def test_pristine_pending_all_paths_and_result_itself_is_delivery_evidence(self):
|
||||
for path in PATHS:
|
||||
with self.subTest(path=path):
|
||||
self.clean(); cid=self.seed(); before=self.row(cid)
|
||||
self.retire(path,cid)
|
||||
self.assert_retired(cid,before,"unknown" if path=="result" else "not_dispatched")
|
||||
|
||||
def test_each_durable_evidence_source_is_unknown_on_every_path(self):
|
||||
sources=("claim","completion","result","error","legacy_result","unlinked_legacy_result","audit_old","audit_new","prior_unknown")
|
||||
for source in sources:
|
||||
for path in PATHS:
|
||||
with self.subTest(source=source,path=path):
|
||||
self.clean(); cid=self.seed()
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
if source in ("claim","completion","result","error"):
|
||||
assignments={"claim":"claimed_at=now()", "completion":"completed_at=now()",
|
||||
"result":"result_payload='{}'::jsonb", "error":"error_message='ambiguous legacy error'"}
|
||||
cur.execute("UPDATE minerops_agent_command_queue SET "+assignments[source]+" WHERE id=%s",(cid,))
|
||||
elif source in ("legacy_result", "unlinked_legacy_result"):
|
||||
cur.execute("""INSERT INTO minerops_agent_command_results (agent_id,site_name,command_id,status,payload)
|
||||
VALUES (%s,'simulated',%s,'completed','{}'::jsonb)""",(AGENT,cid if source=="legacy_result" else None))
|
||||
else:
|
||||
# Durable audit evidence with all queue timestamps missing.
|
||||
old={"command_id":cid,"agent_id":AGENT} if source!="audit_new" else {"agent_id":AGENT}
|
||||
new={"operation_id":cid} if source=="audit_new" else {"execution_outcome":"unknown"}
|
||||
action="agent.command.rejected" if source=="prior_unknown" else "legacy.command.delivered"
|
||||
cur.execute("""INSERT INTO audit_logs (actor_user_id,action,entity_type,old_value_json,new_value_json)
|
||||
VALUES (%s,%s,'agent_command',%s,%s)""",(ACTOR,action,Jsonb(old),Jsonb(new)))
|
||||
prior_audits=self.audits(cid); before=self.row(cid); self.retire(path,cid)
|
||||
after=self.row(cid)
|
||||
self.assertEqual(after["command_status"],"cancelled")
|
||||
self.assertEqual(after["claimed_at"],before["claimed_at"])
|
||||
self.assertEqual(after["result_payload"],before["result_payload"])
|
||||
if before["completed_at"]: self.assertEqual(after["completed_at"],before["completed_at"])
|
||||
audits=self.audits(cid); self.assertEqual(len(audits),len(prior_audits)+1)
|
||||
self.assertTrue(all(a["new_value_json"].get("execution_outcome")=="unknown" for a in audits))
|
||||
|
||||
def test_creation_audit_alone_does_not_report_delivery(self):
|
||||
cid=self.seed()
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO audit_logs (actor_user_id,action,entity_type,new_value_json)
|
||||
VALUES (%s,'miner.control.queued','miner',%s)""",(ACTOR,Jsonb({"operation_id":cid})))
|
||||
before=self.row(cid); self.retire("cutover",cid); self.assert_retired(cid,before,"not_dispatched")
|
||||
|
||||
def test_currently_claimed_without_timestamp_remains_unknown(self):
|
||||
for path in PATHS:
|
||||
self.clean(); cid=self.seed()
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("UPDATE minerops_agent_command_queue SET command_status='claimed' WHERE id=%s",(cid,))
|
||||
before=self.row(cid); self.retire(path,cid); self.assert_retired(cid,before,"unknown")
|
||||
|
||||
def test_audit_failure_rolls_back_every_cancellation_path(self):
|
||||
for path in PATHS:
|
||||
with self.subTest(path=path):
|
||||
self.clean(); cid=self.seed(); before=self.row(cid)
|
||||
def fail(cur,row,**kwargs): cur.execute("SELECT 1/0")
|
||||
with patch.object(AgentCommandRepository,"_audit_rejection",side_effect=fail):
|
||||
with self.assertRaises(Exception): self.retire(path,cid)
|
||||
self.assertEqual(self.row(cid),before); self.assertEqual(self.audits(cid),[])
|
||||
|
||||
def test_concurrent_baseline_claim_blocks_result_then_unknown(self):
|
||||
cid=self.seed(); locker=connect()
|
||||
try:
|
||||
with locker.cursor() as cur:
|
||||
cur.execute(BASELINE_CLAIM_SQL,(AGENT,1)); self.assertEqual(len(cur.fetchall()),1)
|
||||
with ThreadPoolExecutor(max_workers=1) as pool:
|
||||
def cancellation():
|
||||
def named_connection():
|
||||
conn=connect()
|
||||
conn.execute("SET application_name='stage-a-final-result-race'")
|
||||
return conn
|
||||
with self.assertRaises(AgentCommandConflict):
|
||||
AgentCommandRepository(named_connection).finish_command(cid,AGENT,"completed",{},None)
|
||||
future=pool.submit(cancellation)
|
||||
try:
|
||||
# Observe an actual database lock wait before committing the
|
||||
# claim, rather than relying on thread scheduling or sleeps.
|
||||
blocked=False; deadline=time.monotonic()+5
|
||||
while time.monotonic()<deadline:
|
||||
with connect() as observer, observer.cursor() as cur:
|
||||
cur.execute("""SELECT EXISTS (SELECT 1 FROM pg_stat_activity
|
||||
WHERE application_name='stage-a-final-result-race'
|
||||
AND wait_event_type='Lock')""")
|
||||
blocked=cur.fetchone()[0]
|
||||
if blocked: break
|
||||
time.sleep(.01)
|
||||
self.assertTrue(blocked,"Cancellation must wait for the genuine legacy claim lock")
|
||||
finally:
|
||||
locker.commit()
|
||||
future.result(timeout=10)
|
||||
self.assertEqual(self.row(cid)["command_status"],"cancelled")
|
||||
self.assertIsNotNone(self.row(cid)["claimed_at"])
|
||||
self.assertEqual(self.audits(cid)[0]["new_value_json"]["execution_outcome"],"unknown")
|
||||
finally: locker.close()
|
||||
|
||||
def test_mixed_legacy_backlog_classifies_each_row_without_delivering(self):
|
||||
for n in range(12):
|
||||
cid=self.seed("stage-a-final-evidence-"+str(n),"miner_control" if n%2 else "unknown_legacy")
|
||||
if n%3:
|
||||
with connect() as conn,conn.cursor() as cur:
|
||||
cur.execute("UPDATE minerops_agent_command_queue SET claimed_at=now() WHERE id=%s",(cid,))
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
self.assertEqual(sum(pool.map(lambda _: self.repo.retire_unsupported(AGENT,actor_user_id=ACTOR), range(2))),12)
|
||||
for n in range(12):
|
||||
audit=self.audits("stage-a-final-evidence-"+str(n)); self.assertEqual(len(audit),1)
|
||||
self.assertEqual(audit[0]["new_value_json"]["execution_outcome"],"unknown" if n%3 else "not_dispatched")
|
||||
@@ -0,0 +1,309 @@
|
||||
"""Audited schedule retirement on an explicitly disposable PostgreSQL database."""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from unittest.mock import patch
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
from app import main
|
||||
from app.agent_commands.exceptions import AgentCommandConflict
|
||||
from app.agent_commands.repository import AgentCommandRepository
|
||||
from app.agent_commands.policy import PHYSICAL_CONTROL_DISABLED
|
||||
from app.config import settings
|
||||
from app.identity import get_current_user, required_api_permission
|
||||
from tests.test_stage_a_containment_postgres import connect
|
||||
|
||||
|
||||
AGENT = "stage-a-schedule-agent"
|
||||
OTHER_AGENT = "stage-a-schedule-other-agent"
|
||||
ACTOR = UUID("00000000-0000-4000-8000-000000003301")
|
||||
SCHEDULE = "stage-a-schedule-history"
|
||||
|
||||
|
||||
class StageAScheduleCancellationTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if os.environ.get("MINEROPS_STAGE_A_DISPOSABLE") != "1" or not settings.postgres_db.startswith("minerops_stage_a_"):
|
||||
raise RuntimeError("Refusing schedule tests outside an explicitly disposable Stage A database")
|
||||
|
||||
def setUp(self):
|
||||
self.no_transport = patch("socket.create_connection", side_effect=AssertionError("No miner connection"))
|
||||
self.transport = self.no_transport.start()
|
||||
self.addCleanup(self.no_transport.stop)
|
||||
self.addCleanup(self.transport.assert_not_called)
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("DELETE FROM audit_logs WHERE actor_user_id = %s OR old_value_json ->> 'agent_id' IN (%s, %s)",
|
||||
(ACTOR, AGENT, OTHER_AGENT))
|
||||
cur.execute("DELETE FROM minerops_agent_command_queue WHERE agent_id IN (%s, %s)", (AGENT, OTHER_AGENT))
|
||||
cur.execute("DELETE FROM minerops_scheduled_bulk_actions WHERE id = %s", (SCHEDULE,))
|
||||
cur.execute("""INSERT INTO users (id, username, display_name, password_hash, role_id)
|
||||
SELECT %s, 'stage-a-retirement-auditor', 'Retirement Auditor', 'unused', id
|
||||
FROM roles WHERE name = 'Admin' ON CONFLICT (id) DO NOTHING""", (ACTOR,))
|
||||
cur.execute("""INSERT INTO minerops_scheduled_bulk_actions
|
||||
(id, agent_id, name, target_type, scheduled_for, status, last_error)
|
||||
VALUES (%s, %s, 'Historical schedule', 'all', now(), 'running', 'prior evidence')""",
|
||||
(SCHEDULE, AGENT))
|
||||
self.repository = AgentCommandRepository(connect)
|
||||
|
||||
def seed(self, command_id, status="pending", agent=AGENT, kind="miner_control", schedule=SCHEDULE):
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO minerops_agent_command_queue
|
||||
(id, agent_id, command_type, command_status, request_payload, result_payload,
|
||||
error_message, claimed_at)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, 'prior evidence',
|
||||
CASE WHEN %s = 'claimed' THEN now() ELSE NULL END)""",
|
||||
(command_id, agent, kind, status,
|
||||
Jsonb({"scheduled_action_id": schedule, "ip_address": "192.0.2.1", "action": "suspend"}),
|
||||
Jsonb({"historical_result": True}), status))
|
||||
|
||||
@staticmethod
|
||||
def row(command_id):
|
||||
with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("SELECT * FROM minerops_agent_command_queue WHERE id = %s", (command_id,))
|
||||
return dict(cur.fetchone())
|
||||
|
||||
@staticmethod
|
||||
def schedule():
|
||||
with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("SELECT * FROM minerops_scheduled_bulk_actions WHERE id = %s", (SCHEDULE,))
|
||||
return dict(cur.fetchone())
|
||||
|
||||
@staticmethod
|
||||
def audits():
|
||||
with connect() as conn, conn.cursor(row_factory=dict_row) as cur:
|
||||
cur.execute("""SELECT * FROM audit_logs WHERE actor_user_id = %s
|
||||
OR old_value_json ->> 'agent_id' IN (%s, %s)""", (ACTOR, AGENT, OTHER_AGENT))
|
||||
return [dict(row) for row in cur.fetchall()]
|
||||
|
||||
def retire(self, operation="cancel"):
|
||||
with patch.object(main, "get_db_connection", side_effect=connect):
|
||||
return getattr(main, f"{operation}_dashboard_scheduled_bulk_action")(
|
||||
SCHEDULE, {"id": str(ACTOR), "username": "retirement-auditor"},
|
||||
)
|
||||
|
||||
def assert_retired(self, command_id, before, operation):
|
||||
after = self.row(command_id)
|
||||
self.assertEqual(after["command_status"], "cancelled")
|
||||
for field in ("id", "agent_id", "command_type", "request_payload", "result_payload", "created_at", "claimed_at"):
|
||||
self.assertEqual(after[field], before[field], field)
|
||||
self.assertEqual(after["error_message"], "prior evidence\n" + PHYSICAL_CONTROL_DISABLED)
|
||||
self.assertIsNotNone(after["completed_at"])
|
||||
audits = [r for r in self.audits() if r["action"] == "agent.command.rejected"
|
||||
and r["old_value_json"]["command_id"] == command_id]
|
||||
self.assertEqual(len(audits), 1)
|
||||
audit = audits[0]
|
||||
self.assertEqual(audit["actor_user_id"], ACTOR)
|
||||
self.assertEqual(audit["old_value_json"]["agent_id"], AGENT)
|
||||
self.assertEqual(audit["old_value_json"]["status"], before["command_status"])
|
||||
self.assertEqual(audit["new_value_json"]["source"], "schedule_" + operation)
|
||||
self.assertEqual(audit["new_value_json"]["reason"], PHYSICAL_CONTROL_DISABLED)
|
||||
self.assertEqual(audit["new_value_json"]["execution_outcome"],
|
||||
"unknown")
|
||||
self.assertNotIn("192.0.2.1", str(audit))
|
||||
|
||||
def test_schedule_cancel_retires_pending_and_claimed_with_attributed_audits(self):
|
||||
for status in ("pending", "claimed"):
|
||||
command_id = "stage-a-schedule-cancel-" + status
|
||||
self.seed(command_id, status)
|
||||
before = {s: self.row("stage-a-schedule-cancel-" + s) for s in ("pending", "claimed")}
|
||||
self.retire()
|
||||
for status in before:
|
||||
self.assert_retired("stage-a-schedule-cancel-" + status, before[status], "cancel")
|
||||
self.assertEqual(self.schedule()["status"], "cancelled")
|
||||
self.assertEqual(len(self.audits()), 3)
|
||||
|
||||
def test_schedule_delete_is_audited_retirement_and_preserves_history(self):
|
||||
self.seed("stage-a-schedule-delete", "claimed")
|
||||
before, schedule_before = self.row("stage-a-schedule-delete"), self.schedule()
|
||||
self.retire("delete")
|
||||
self.assert_retired("stage-a-schedule-delete", before, "delete")
|
||||
after = self.schedule()
|
||||
for field in schedule_before:
|
||||
if field not in {"status", "last_error", "updated_at"}:
|
||||
self.assertEqual(after[field], schedule_before[field], field)
|
||||
parent_audits = [r for r in self.audits() if r["action"] == "scheduled_action.retired"]
|
||||
self.assertEqual(len(parent_audits), 1)
|
||||
self.assertEqual(parent_audits[0]["new_value_json"]["operation"], "delete")
|
||||
|
||||
def test_repeated_cancel_and_delete_are_idempotent(self):
|
||||
self.seed("stage-a-schedule-repeat")
|
||||
self.retire()
|
||||
before, schedule_before, audits_before = self.row("stage-a-schedule-repeat"), self.schedule(), self.audits()
|
||||
for operation in ("cancel", "delete", "delete"):
|
||||
self.retire(operation)
|
||||
self.assertEqual(self.row("stage-a-schedule-repeat"), before)
|
||||
self.assertEqual(self.schedule(), schedule_before)
|
||||
self.assertEqual(self.audits(), audits_before)
|
||||
|
||||
def test_terminal_queue_and_schedule_history_remains_immutable(self):
|
||||
for status in ("completed", "failed", "cancelled"):
|
||||
self.seed("stage-a-schedule-terminal-" + status, status)
|
||||
before = {s: self.row("stage-a-schedule-terminal-" + s) for s in ("completed", "failed", "cancelled")}
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("UPDATE minerops_scheduled_bulk_actions SET status = 'completed' WHERE id = %s", (SCHEDULE,))
|
||||
schedule_before = self.schedule()
|
||||
self.retire("delete")
|
||||
self.assertEqual(self.schedule(), schedule_before)
|
||||
for status, row in before.items():
|
||||
self.assertEqual(self.row("stage-a-schedule-terminal-" + status), row)
|
||||
self.assertEqual(self.audits(), [])
|
||||
|
||||
def test_stored_agent_and_schedule_scope_cannot_be_overridden(self):
|
||||
self.seed("stage-a-schedule-own")
|
||||
self.seed("stage-a-schedule-foreign", agent=OTHER_AGENT)
|
||||
self.seed("stage-a-schedule-unrelated", schedule="another-schedule")
|
||||
self.seed("stage-a-schedule-telemetry", kind="poll_known_miners_now")
|
||||
protected = {c: self.row(c) for c in ("stage-a-schedule-foreign", "stage-a-schedule-unrelated", "stage-a-schedule-telemetry")}
|
||||
self.retire()
|
||||
self.assertEqual(self.row("stage-a-schedule-own")["command_status"], "cancelled")
|
||||
for command_id, before in protected.items():
|
||||
self.assertEqual(self.row(command_id), before)
|
||||
|
||||
def test_schedule_unknown_type_is_rejected_with_its_actual_reason(self):
|
||||
self.seed("stage-a-schedule-unknown", kind="raw_api3")
|
||||
self.retire()
|
||||
rejection = next(a for a in self.audits() if a["action"] == "agent.command.rejected")
|
||||
self.assertEqual(rejection["new_value_json"]["reason"], "unsupported_command_schema_stage_a")
|
||||
|
||||
def test_concurrent_cancel_delete_workers_audit_each_transition_once(self):
|
||||
self.seed("stage-a-schedule-concurrent", "claimed")
|
||||
def worker(operation):
|
||||
return AgentCommandRepository(connect).retire_schedule(SCHEDULE, actor_user_id=str(ACTOR), operation=operation)
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
counts = list(pool.map(worker, ("cancel", "delete")))
|
||||
self.assertEqual(sorted(counts), [0, 1])
|
||||
self.assertEqual(self.row("stage-a-schedule-concurrent")["command_status"], "cancelled")
|
||||
self.assertEqual(len(self.audits()), 2)
|
||||
|
||||
def test_poll_finish_and_schedule_cancel_race_has_one_queue_audit(self):
|
||||
self.seed("stage-a-schedule-race", "claimed")
|
||||
def worker(i):
|
||||
repo = AgentCommandRepository(connect)
|
||||
try:
|
||||
if i == 0:
|
||||
return repo.retire_schedule(SCHEDULE, actor_user_id=str(ACTOR), operation="cancel")
|
||||
if i == 1:
|
||||
return repo.claim_pending(AGENT)
|
||||
return repo.finish_command("stage-a-schedule-race", AGENT, "completed", {}, None)
|
||||
except AgentCommandConflict:
|
||||
return None
|
||||
with ThreadPoolExecutor(max_workers=3) as pool:
|
||||
list(pool.map(worker, range(3)))
|
||||
rejections = [a for a in self.audits() if a["action"] == "agent.command.rejected"]
|
||||
self.assertEqual(len(rejections), 1)
|
||||
self.assertEqual(rejections[0]["new_value_json"]["execution_outcome"], "unknown")
|
||||
self.assertEqual(self.row("stage-a-schedule-race")["command_status"], "cancelled")
|
||||
|
||||
def test_audit_failure_after_one_insert_rolls_back_all_cancellations(self):
|
||||
for command_id in ("stage-a-schedule-rollback-a", "stage-a-schedule-rollback-b"):
|
||||
self.seed(command_id, "claimed")
|
||||
before = {c: self.row(c) for c in ("stage-a-schedule-rollback-a", "stage-a-schedule-rollback-b")}
|
||||
schedule_before = self.schedule()
|
||||
original = AgentCommandRepository._audit_rejection
|
||||
calls = []
|
||||
def fail_second(cur, row, **kwargs):
|
||||
calls.append(row["id"])
|
||||
if len(calls) == 2:
|
||||
raise RuntimeError("simulated audit failure")
|
||||
original(cur, row, **kwargs)
|
||||
with patch.object(AgentCommandRepository, "_audit_rejection", side_effect=fail_second):
|
||||
with self.assertRaisesRegex(RuntimeError, "simulated audit failure"):
|
||||
self.retire("delete")
|
||||
self.assertEqual(len(calls), 2)
|
||||
self.assertEqual(self.schedule(), schedule_before)
|
||||
for command_id, row in before.items():
|
||||
self.assertEqual(self.row(command_id), row)
|
||||
self.assertEqual(self.audits(), [])
|
||||
|
||||
def test_schedule_parent_audit_failure_rolls_back_queue_and_parent(self):
|
||||
self.seed("stage-a-schedule-parent-rollback")
|
||||
before, parent_before = self.row("stage-a-schedule-parent-rollback"), self.schedule()
|
||||
# A database constraint failure after child audits exercises the final
|
||||
# schedule/audit portion, rather than only an early application exception.
|
||||
def connection_with_failure():
|
||||
conn = connect()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("""CREATE OR REPLACE FUNCTION pg_temp.reject_schedule_audit() RETURNS trigger
|
||||
LANGUAGE plpgsql AS $$ BEGIN
|
||||
IF NEW.action = 'scheduled_action.retired' THEN RAISE EXCEPTION 'simulated parent audit failure'; END IF;
|
||||
RETURN NEW; END $$""")
|
||||
cur.execute("CREATE TRIGGER stage_a_parent_audit_failure BEFORE INSERT ON audit_logs FOR EACH ROW EXECUTE FUNCTION pg_temp.reject_schedule_audit()")
|
||||
# Trigger creation is in the very transaction the repository must roll back.
|
||||
return conn
|
||||
with self.assertRaisesRegex(Exception, "simulated parent audit failure"):
|
||||
AgentCommandRepository(connection_with_failure).retire_schedule(
|
||||
SCHEDULE, actor_user_id=str(ACTOR), operation="cancel")
|
||||
self.assertEqual(self.row("stage-a-schedule-parent-rollback"), before)
|
||||
self.assertEqual(self.schedule(), parent_before)
|
||||
self.assertEqual(self.audits(), [])
|
||||
|
||||
def test_unknown_schedule_has_no_queue_or_audit_side_effects(self):
|
||||
self.seed("stage-a-schedule-orphan")
|
||||
before = self.row("stage-a-schedule-orphan")
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("DELETE FROM minerops_scheduled_bulk_actions WHERE id = %s", (SCHEDULE,))
|
||||
from fastapi import HTTPException
|
||||
with self.assertRaises(HTTPException) as error:
|
||||
self.retire()
|
||||
self.assertEqual(error.exception.status_code, 404)
|
||||
self.assertEqual(self.row("stage-a-schedule-orphan"), before)
|
||||
self.assertEqual(self.audits(), [])
|
||||
|
||||
def test_http_cancel_delete_require_control_permission(self):
|
||||
app = FastAPI()
|
||||
for operation in ("cancel", "delete"):
|
||||
path = f"/api/dashboard/scheduled-bulk-actions/{{action_id}}/{operation}"
|
||||
app.add_api_route(path, getattr(main, f"{operation}_dashboard_scheduled_bulk_action"), methods=["POST"])
|
||||
self.assertEqual(required_api_permission("POST", path), "miners.control")
|
||||
app.dependency_overrides[get_current_user] = lambda: {"id": str(ACTOR), "role": "Custom", "permissions": ["agents.manage"]}
|
||||
with TestClient(app) as client, patch.object(main, "get_db_connection") as db:
|
||||
for operation in ("cancel", "delete"):
|
||||
self.assertEqual(client.post(f"/api/dashboard/scheduled-bulk-actions/{SCHEDULE}/{operation}").status_code, 403)
|
||||
db.assert_not_called()
|
||||
|
||||
def test_offline_cutover_batch_is_audited_and_never_claims_readers(self):
|
||||
with connect() as conn, conn.cursor() as cur:
|
||||
cur.execute("""INSERT INTO minerops_agent_command_queue
|
||||
(id, agent_id, command_type, command_status, request_payload, claimed_at)
|
||||
SELECT 'stage-a-schedule-backlog-' || n, %s,
|
||||
CASE WHEN n %% 2 = 0 THEN 'miner_control' ELSE 'raw_api3' END,
|
||||
CASE WHEN n %% 3 = 0 THEN 'claimed' ELSE 'pending' END,
|
||||
'{}'::jsonb, CASE WHEN n %% 5 = 0 THEN now() ELSE NULL END
|
||||
FROM generate_series(1,1003) n""", (AGENT,))
|
||||
self.seed("stage-a-schedule-offline-read", kind="poll_known_miners_now")
|
||||
self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 1000)
|
||||
self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 3)
|
||||
self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 0)
|
||||
self.assertEqual(self.row("stage-a-schedule-offline-read")["command_status"], "pending")
|
||||
rejections = self.audits()
|
||||
self.assertEqual(len(rejections), 1003)
|
||||
for audit in rejections:
|
||||
self.assertEqual(audit["actor_user_id"], ACTOR)
|
||||
self.assertEqual(audit["new_value_json"]["source"], "stage_a_cutover")
|
||||
n = int(audit["old_value_json"]["command_id"].rsplit("-", 1)[1])
|
||||
self.assertEqual(audit["new_value_json"]["execution_outcome"],
|
||||
"unknown" if n % 3 == 0 or n % 5 == 0 else "not_dispatched")
|
||||
|
||||
def test_cutover_zero_batch_does_not_claim_a_locked_drain_is_complete(self):
|
||||
self.seed("stage-a-schedule-cutover-locked", "claimed")
|
||||
locker = connect()
|
||||
try:
|
||||
with locker.cursor() as cur:
|
||||
cur.execute("SELECT id FROM minerops_agent_command_queue WHERE id = %s FOR UPDATE", ("stage-a-schedule-cutover-locked",))
|
||||
self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 0)
|
||||
self.assertEqual(self.row("stage-a-schedule-cutover-locked")["command_status"], "claimed")
|
||||
finally:
|
||||
locker.rollback()
|
||||
locker.close()
|
||||
self.assertEqual(self.repository.retire_unsupported(AGENT, actor_user_id=str(ACTOR)), 1)
|
||||
self.assertEqual(self.row("stage-a-schedule-cutover-locked")["command_status"], "cancelled")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,219 @@
|
||||
Blocker #1 discovery isolation: candidate review and compatibility contract
|
||||
=======================================================================
|
||||
Review candidate only. Blocker #1 is NOT declared PASS. No production cutover is
|
||||
authorized. Base: e898f283ab0fc21bac77b63db1c4a9ad35e21562 (Stage A / draft PR #6).
|
||||
No schema migration, new constraints, backfill, or runtime DDL is introduced.
|
||||
|
||||
Authority and observations
|
||||
-------------------------
|
||||
All machine discovery/telemetry routes retain per-Agent HTTP authentication.
|
||||
Authority comes from the persisted Agent -> Site/Zone/network assignment, never
|
||||
client Site, room, Zone, target-Agent names, or a body-supplied Agent identity.
|
||||
Conflicting body Agent IDs are rejected. Disabled/unassigned/revoked routing fails
|
||||
closed. Check-in cannot establish or change an Agent's initial assignment: use
|
||||
existing authenticated enrollment or the separately reviewed legacy transition.
|
||||
Unassigned inventory is not implicitly claimed by check-in or snapshots.
|
||||
|
||||
Discovery reports are bounded (128 items, 1 MiB body; 1,024 observations/minute per
|
||||
Agent). IP literals must be in the current Zone networks. MACs must be canonical
|
||||
ASCII unicast addresses. Explicit IDs, IP aliases, MAC aliases and supplied serial
|
||||
must agree with current inventory and immutable historical MAC ownership. Global
|
||||
historical MAC/serial collision checks reveal no foreign inventory IDs. Duplicate
|
||||
current IPs within a Zone are conflicts; distinct Zones may use the same IP.
|
||||
|
||||
Each observation has a stable server ID and version. Raw reports and original
|
||||
provenance are retained, with an atomic discovery.observed audit recording the
|
||||
current authority, raw hash, inventory fingerprint and observation version. Exact
|
||||
retries deduplicate; discovery clients must send discovered_at for a fresh scan
|
||||
(the supplied Windows scanner does). Approval requires a current, proven,
|
||||
non-superseded observation less than 15 minutes old. Legacy/null/mislinked rows
|
||||
without this evidence remain visible but cannot authorize approval or enrichment.
|
||||
There is no automatic evidence backfill, deletion, identity merge or ownership
|
||||
transfer. An authorized operator must investigate conflicts and obtain a fresh
|
||||
observation. Discovery does not implement an ownership-transfer workflow.
|
||||
|
||||
Human policy remains global administration: agents.view for discovery display;
|
||||
agents.manage for stable-ID review/enrichment; miners.control for scan/poll
|
||||
requests. The UI displays authoritative Agent/Zone and escapes all untrusted text.
|
||||
IP-only approval is unsupported. Submit discovery_id, version and one of
|
||||
add_to_inventory / ignore / non_miner. All three revalidate current authority and
|
||||
identity under locks; only that observation changes. Adding new inventory also
|
||||
requires a valid MAC. Repeat the same accepted decision/version to receive its
|
||||
recorded response without another audit or insertion. Stale/different decisions
|
||||
conflict; refresh before retrying. Audit actor is the authenticated human.
|
||||
|
||||
Read commands and enrichment
|
||||
----------------------------
|
||||
Enrichment is selected by discovery_id/version, routed to its actual observing
|
||||
Agent, and bound to the current authority and IP. No arbitrary-IP or hardcoded
|
||||
Plexus fallback exists. Scan/poll requests require an explicit registered Agent ID;
|
||||
optional UUID request_id provides explicit retry idempotency. New requests without
|
||||
request_id receive distinct IDs. Queue claim rejects legacy/unscoped/stale reads
|
||||
with atomic cancellation/audit and preserves unknown delivery outcome for any
|
||||
previous claim evidence. Existing Stage A physical rejection remains in force.
|
||||
|
||||
Before enrichment miner I/O, the upgraded Windows child submits command_id,
|
||||
discovery_id and version to authenticated POST /api/agent/discovery-target. Only
|
||||
a current claimed enrichment command returns a canonical IP. Enrichment submission
|
||||
revalidates the same binding and appends a NEW child observation for human review;
|
||||
it never resets the parent's decision or reparents history. V1/V2/V3 share the
|
||||
same boundary. Replaying old V2 identity merge inputs cannot merge inventory.
|
||||
|
||||
Known-miners returns a short-lived signed miner_version bound to Agent authority
|
||||
and the current miner ID/IP/MAC/updated_at fingerprint. Every miner/pool/API3
|
||||
snapshot or known-miner enrichment must carry this token and matching identifiers.
|
||||
All Windows polling clients propagate it and batch at 128. A concurrent update,
|
||||
revocation or reassignment may invalidate in-flight telemetry: discard the stale
|
||||
response, refetch known-miners and poll again. Do not fall back to IP matching.
|
||||
Descriptive enrichment does not change MAC/IP/serial/ownership. Raw diagnostics
|
||||
are retained in append-only audits, not empty status rows that hide measurements.
|
||||
API3 retains raw snapshots, safe descriptive metadata, actual summary/pool/board
|
||||
measurements and compatibility counters. Failed/metadata-only API3 responses do
|
||||
not replace the most recent measurements. Nested identity contradictions fail.
|
||||
|
||||
Transactions and database roles
|
||||
------------------------------
|
||||
All discovery writers/read-command claims take bounded locks in a shared order:
|
||||
advisory transaction lock 743190031, SHARE on Agent/Zone/network tables, SHARE ROW
|
||||
EXCLUSIVE on miners, then row locks. These table locks prevent phantom inserts
|
||||
from legacy writers which do not use the advisory lock. Agent routing/check-in
|
||||
uses guarded updates. No miner network I/O occurs while these transactions run.
|
||||
Audit insertion, inventory mutation and the observation decision commit together;
|
||||
any error rolls back all changes. Audit-denial and real UPDATE-trigger failure,
|
||||
concurrent approvals, cross-Zone MAC races, Zone disable and claim-vs-review are
|
||||
covered with real PostgreSQL connections and registered HTTP authorization routes.
|
||||
Lock timeout is 5 seconds; statement timeout 15 seconds. Controlled contention
|
||||
returns conflict/retry. Security favors correctness over fleet throughput here.
|
||||
|
||||
Required release review: validate privileges/lock behavior for the actual deployed
|
||||
DB role in a disposable replica and measure expected fleet load. Tests use an
|
||||
isolated database owner plus a limited role with audit INSERT denied. They do not
|
||||
certify the production role. No new uniqueness constraint is proposed, so do not
|
||||
remove the transaction/table locks without a separately reviewed replacement.
|
||||
|
||||
Reproduction
|
||||
------------
|
||||
Use a disposable checkout of this candidate with an already available backend
|
||||
image containing backend/requirements.txt. Never mount production credentials or
|
||||
.env files. From the candidate root:
|
||||
|
||||
MINEROPS_REVIEW_BACKEND_IMAGE=<existing-image> scripts/test-discovery-security.sh
|
||||
|
||||
The runner creates a unique internal Docker network, tmpfs PostgreSQL 16, no
|
||||
published ports, applies the unchanged existing Alembic chain and runs the two
|
||||
PostgreSQL discovery suites. A trap destroys only its own resources. Tests use
|
||||
simulated Agents and FastAPI's registered routes, without app startup/schedulers.
|
||||
|
||||
PowerShell (isolated PowerShell 7 host/container with simulated/loopback I/O):
|
||||
pwsh -NoProfile -File agents/windows/tests/Test-MinerOpsDiscovery.ps1
|
||||
pwsh -NoProfile -File agents/windows/tests/Test-MinerOpsControl.ps1
|
||||
pwsh -NoProfile -File agents/windows/tests/Test-MinerOpsDiagnosticTransport.ps1
|
||||
|
||||
Browser (Python 3, Node, Playwright and its Chromium/runtime installed externally):
|
||||
python3 backend/tests/browser/export_discovery_page.py . /tmp/discovery-page.html
|
||||
node backend/tests/browser/discovery-security.cjs /tmp/discovery-page.html
|
||||
|
||||
The browser test intercepts all requests, supplies harmless event-marker input,
|
||||
and verifies inert rendering and exact same-IP discovery_id/version selection.
|
||||
Its trailing baseline argument demonstrates the baseline stored-markup defect.
|
||||
|
||||
Other regression suites (explicit disposable DB guards apply):
|
||||
tests.test_stage_a_containment_postgres
|
||||
tests.test_stage_a_delivery_evidence_postgres
|
||||
tests.test_stage_a_schedule_cancellation_postgres
|
||||
tests.test_agent_zones_postgres
|
||||
tests.test_agent_credential_transition_postgres
|
||||
tests.test_agent_enrollment_postgres
|
||||
|
||||
Baseline source-test exceptions retained unchanged
|
||||
-------------------------------------------------
|
||||
backend/tests/test_agent_commands.py:
|
||||
AgentEnrichmentIsolationSourceTests.test_legacy_config_remains_explicitly_legacy_only
|
||||
requires the old minerops_require_agent_secret source string / fixed Plexus config.
|
||||
The authenticated config route already changed before this candidate. Replace the
|
||||
obsolete source assertion with real legacy/authenticated route tests in separate
|
||||
work; do not weaken credential authentication to satisfy the string assertion.
|
||||
|
||||
AgentSourceContractTests.test_deployment_preserves_local_state_and_registers_established_tasks
|
||||
blanket-bans the quoted agent-config.json filename in installer source, although
|
||||
the existing credential-recovery installer legitimately preserves/recovers local
|
||||
configuration. Follow up with behavioral installation/state-preservation tests on
|
||||
Windows. Do not remove the recovery path to satisfy the blanket source assertion.
|
||||
|
||||
Both failures reproduce on unchanged 2d89e80 and unchanged e898f28; both method
|
||||
bodies remain identical. They are not silent skips or weakened assertions here.
|
||||
The external-fixture legacy-retirement PostgreSQL test class also requires its
|
||||
own migration-case setup and is not certified by the discovery test runner.
|
||||
|
||||
Outstanding gates
|
||||
-----------------
|
||||
Independent security review of this exact isolated candidate is required before
|
||||
any Blocker #1 PASS or commit approval. Native Windows PowerShell 5.1/SYSTEM and
|
||||
fleet-scale/production-role validation remain outstanding. Unsupported older
|
||||
Agent payloads fail closed and require coordinated upgrade; this is an intentional
|
||||
protocol compatibility change, not transparent mixed-version support.
|
||||
|
||||
Stage A application containment remains a separate audited boundary. Physical
|
||||
execution is disabled; device identity binding and Blocker #3 Stages B-E remain
|
||||
outstanding. Preserve STAGE_A_OLD_AGENT_CUTOVER.txt: verified network fencing must
|
||||
precede stopping/draining/upgrading old Agents. Cached commands cannot be revoked
|
||||
by server cancellation alone. Offline/unreachable Agents stay blocked; restart,
|
||||
reconnection and rollback must not restore an unsafe executor. This candidate
|
||||
neither performs nor authorizes that cutover or any production deployment.
|
||||
|
||||
Independent-audit remediation addendum (October 16, 2026)
|
||||
-------------------------------------------------------
|
||||
The original candidate (99f515924dfc85fe1e1e98361688f1060329c090) failed independent
|
||||
review for F1-F3 in retained legacy callers. This corrected candidate requires a
|
||||
new independent audit. It does not claim Blocker #1 PASS.
|
||||
|
||||
Retired in Hub AND non-Hub modes, with HTTP 410 after existing authentication and
|
||||
permission checks:
|
||||
POST /api/miners/discover-one
|
||||
POST /api/scan-ranges/{scan_range_id}/discover-now
|
||||
POST /api/scanner/run-known-now
|
||||
POST /api/tools/legacy-summary-refresh-one
|
||||
POST /api/tools/legacy-summary-refresh-missing
|
||||
All definitions/aliases of their identity matcher, discovery writer, scheduler
|
||||
adapters, summary writer and global-IP helper reject before transport/database
|
||||
access. Startup creates no local scanner thread; status reports retired/disabled.
|
||||
Ordinary miners.update therefore cannot use discovery to replace a control board
|
||||
or clear a MAC. The dedicated board workflow and immutable MAC ownership remain.
|
||||
|
||||
Functional implication: legacy local discovery, automatic local refresh, and
|
||||
legacy one/bulk summary backfill are unavailable even outside Hub mode. Operators
|
||||
must use authenticated Agent observation/review and inventory-bound Agent polling.
|
||||
Legacy summary replies lack trustworthy identity; there is no retained unversioned
|
||||
write or in-flight read to revalidate. No denied request mutates inventory, audit,
|
||||
or telemetry. The new shared discovery architecture is preserved.
|
||||
|
||||
CSV remains an explicit global administrative workflow protected by miners.update.
|
||||
Each row requires Miner ID, current MAC, serial, or IP plus Agent Zone ID. Zone may
|
||||
come from the query/UI input or a row column; conflicting selections are rejected.
|
||||
All supplied ID/MAC/serial/IP/Site/Zone evidence must agree with current inventory.
|
||||
A failed MAC/serial/ID never falls back to global IP. Missing inventory is reported
|
||||
unmatched; contradictory or ambiguous selection returns row-specific 409, malformed
|
||||
or insufficient identification 422. A failure rolls back the entire import.
|
||||
IP-only rows need a Zone even when only one global match currently exists. Exact
|
||||
MAC/ID/serial selection remains usable by global administrators across Sites.
|
||||
CSV updates only model/type/firmware, never ownership, IP, MAC or telemetry.
|
||||
Duplicate columns are rejected rather than silently using the last value.
|
||||
|
||||
CSV acquires the existing discovery lock order, including the miners table lock,
|
||||
before resolving any row. It holds the locks through metadata changes and one
|
||||
atomic audit containing every changed inventory ID/namespace and old/new values.
|
||||
Lock contention is bounded by existing discovery timeouts; large fleet throughput
|
||||
and actual deployed-role privileges still require independent release validation.
|
||||
Preview uses identical identification checks and rolls back. Repeated imports do
|
||||
not create inventory or telemetry; each applied request retains its import audit.
|
||||
No migration or containment/control policy changes are part of this remediation.
|
||||
|
||||
Regression entrypoint: scripts/test-discovery-security.sh now includes
|
||||
tests.test_legacy_inventory_security_postgres
|
||||
with registered HTTP requests, actual PostgreSQL locks/concurrent ownership moves,
|
||||
actual audit INSERT denial rollback, retired helper/alias coverage, both modes,
|
||||
permission boundaries, ambiguous/conflicting CSV selection, legitimate imports,
|
||||
preview/replay, and retained historical snapshots. All devices are simulated or
|
||||
transport-patched; the runner uses an internal network and disposable database.
|
||||
Full execution evidence and integrity manifest are retained outside the candidate
|
||||
in the remediation evidence directory referenced by the final report.
|
||||
@@ -0,0 +1,248 @@
|
||||
MinerOps Blocker #3 Stage A: old-Agent containment and upgrade runbook
|
||||
Status: review candidate. This procedure has NOT been executed in production.
|
||||
Scope: contain physical control; no Stages B-E protocol or control re-enablement.
|
||||
|
||||
The independent audit demonstrated that an old Agent can execute a previously
|
||||
delivered command after the Hub marks it cancelled. Cancelling a database row,
|
||||
revoking a Hub credential, advertising control unavailable, or replacing script
|
||||
files does not revoke a cached command or loaded PowerShell functions. Execution
|
||||
of any previously claimed command remains UNKNOWN until separately investigated.
|
||||
|
||||
Production approval requires independent security revalidation, an approved change
|
||||
window, verified network enforcement, and the native Windows checks below. A new
|
||||
database migration or production network infrastructure change requires explicit
|
||||
approval. This runbook authorizes no production actions by itself.
|
||||
|
||||
Ownership and evidence
|
||||
|
||||
Assign an incident/change owner, a network operator, a Hub/database operator, and a
|
||||
Windows operator. Record their identities, UTC timestamps, change ticket, and every
|
||||
gate's evidence in a protected location. Keep enrollment tokens and Agent/miner
|
||||
credentials out of ordinary logs and this checklist. Preserve required backups
|
||||
and recovery state with their existing ACLs; do not overwrite agent-config.json.
|
||||
|
||||
Maintain one ledger containing ALL registered Agents, every distinct queue Agent
|
||||
(including deleted registry entries), offline/unreachable Agents, alternative
|
||||
installations and manually running executors, legacy shared identities, all Hub
|
||||
replicas/workers, scheduled jobs, task hosts, physical networks, and IPv4/IPv6 paths.
|
||||
For each Agent record its Site/Zone assignment, last check-in, host/network identity,
|
||||
network fence evidence, process/task evidence, installed-file hashes, verified
|
||||
artifact identifier, and release decision. Keep unverified entries quarantined.
|
||||
|
||||
Gate 1: restrict physical write access BEFORE stopping/upgrading anything
|
||||
|
||||
1. Arrange an approved maintenance window and identify the actual network control
|
||||
that can keep old Agents from reaching ALL physical miners. Layer-4 rules cannot
|
||||
distinguish a legacy read from a write on the same miner API port. Default to
|
||||
denying Agent-to-miner traffic, including ports 4028 and 4433, alternative miner
|
||||
management interfaces/ports, IPv6, alternate NICs/VPNs, and established sessions.
|
||||
Telemetry may be interrupted. Do not substitute an unreviewed 'read-only' proxy.
|
||||
2. Apply only separately approved network controls through the responsible operator.
|
||||
A stopped process or a host-local rule alone is insufficient for an unreachable
|
||||
machine that can later reboot/reconnect. Use enforcement that already applies
|
||||
to that machine's traffic when it returns. Include unknown/retired installations
|
||||
and aliases; trusting an IP address alone is insufficient when addresses can change.
|
||||
3. Verify deny counters and session termination/conntrack handling. Use approved
|
||||
synthetic targets/canaries, never a real write command to a physical miner.
|
||||
A rule that blocks only NEW connections leaves a pre-existing API session exposed.
|
||||
Check both the intended route and alternate routes. Keep the fence if evidence
|
||||
is unavailable, ambiguous, or an offline Agent cannot be located.
|
||||
|
||||
Exit gate: verified deny enforcement covers every unverified/legacy executor,
|
||||
including future reconnects and existing sessions. Failure means STOP; keep fences.
|
||||
|
||||
Gate 2: stop old executors and automatic restarts
|
||||
|
||||
1. Stop command production and processing on every old Hub/worker version, including
|
||||
old schedule materializers and running requests that could still insert/deliver
|
||||
commands. Hold ingress and scale/restart mechanisms as necessary under the
|
||||
approved change procedure. Inventory the actual versions of every instance.
|
||||
2. On each reachable Windows host disable/hold all relevant Scheduled Task triggers,
|
||||
retry/restart policies, service recovery, watchdogs, startup/logon entries, GUI
|
||||
launch actions, manual jobs, and external orchestration that can launch MinerOps.
|
||||
Record original definitions and intended running/stopped state for recovery.
|
||||
3. Stop MinerOpsAgent and any alternative/manual executor, then verify task instances
|
||||
AND their PowerShell/child process trees have exited. Record command paths,
|
||||
process start times and identities. An in-memory old module is still executable
|
||||
even after its source file changes. Do not gracefully 'drain' by executing writes.
|
||||
4. Keep pool/enrichment tasks held during runtime replacement and verify no alternate
|
||||
old code copy can auto-start. Never run uninstall or erase config/state to stop it.
|
||||
|
||||
Exit gate: every reachable old executor is stopped and restart-held; every
|
||||
unreachable/unverified Agent remains network-fenced. Do not remove a fence to
|
||||
diagnose a failed stop. Escalate failures with evidence and keep maintenance holds.
|
||||
|
||||
Gate 3: update ALL Hub instances with no mixed-version execution
|
||||
|
||||
1. Capture protected queue/audit evidence and the relevant schedule/Agent inventory
|
||||
before changes. Retain original request/result payloads, timestamps and errors.
|
||||
Secure these exports because historical payloads may contain sensitive data.
|
||||
2. Install only the independently reviewed Stage A artifact on every Hub instance
|
||||
while executor fences/holds remain. Stop old replicas and workers fully, including
|
||||
rolling-update stragglers and old connections/requests; do not allow a mixture of
|
||||
active baseline and Stage A schedulers/queue consumers.
|
||||
3. Verify each instance's actual artifact hashes and that physical admission is
|
||||
fixed disabled: legacy write/diagnostic POSTs return 410 after authorization,
|
||||
canonical typed controls return 409 physical_control_disabled_stage_a, and queue
|
||||
SQL dispatches only the fixed read types. Do not use runtime flags to enable writes.
|
||||
4. Run reviewed schema compatibility checks as appropriate. Stage A remediation adds
|
||||
NO migration. Stop and request approval if the deployed schema needs one.
|
||||
|
||||
Exit gate: only reviewed containment Hub code is active; all old producers are
|
||||
stopped. Keep Agents held. If any instance cannot be verified, keep ingress/worker
|
||||
holds and physical fences rather than continuing a partial rolling deployment.
|
||||
|
||||
Gate 4: audited cancellation across ALL Agents, including offline/queue-only IDs
|
||||
|
||||
1. Use the canonical reviewed code and a real accountable operator users.id for
|
||||
manual cutover retirement. Existing authenticated Agent polling/result rejection
|
||||
is system-attributed by source and Agent ID; manual cutover supplies actor_user_id.
|
||||
Do not impersonate another user, create invented execution evidence, or log secrets.
|
||||
2. Inventory IDs using the UNION of registry and queue Agent IDs. Also reconcile the
|
||||
fleet ledger, historical/retired identities, and old shared Agent ID. Do not use
|
||||
only Agents currently checking in. A useful read-only inventory is:
|
||||
|
||||
SELECT agent_id FROM minerops_site_agents
|
||||
UNION SELECT agent_id FROM minerops_agent_command_queue
|
||||
ORDER BY agent_id;
|
||||
|
||||
3. In a separately approved administrative process configured for the intended DB,
|
||||
call the reviewed repository operation for each ID, in committed bounded batches:
|
||||
|
||||
repository = AgentCommandRepository()
|
||||
count = repository.retire_unsupported(agent_id, actor_user_id=operator_user_id)
|
||||
|
||||
Repeat for all IDs while progress is made, under an explicit time/batch limit.
|
||||
This method never claims read work; do NOT substitute claim_pending or an Agent
|
||||
HTTP poll as an administrative drain. Each transition and rejection audit commits
|
||||
atomically. Database/audit failure rolls back and keeps the change window closed.
|
||||
4. A batch is bounded at 1000 and uses SKIP LOCKED. A zero count may mean locked rows,
|
||||
not completion. Resolve/retry approved lock holders with time limits; never
|
||||
silently declare the drain finished. Query ALL Agents to verify a zero live count:
|
||||
|
||||
SELECT agent_id, command_type, command_status, count(*)
|
||||
FROM minerops_agent_command_queue
|
||||
WHERE command_status IN ('pending', 'claimed')
|
||||
AND command_type NOT IN
|
||||
('poll_known_miners_now', 'discover_new_miners', 'enrich_match_discovery')
|
||||
GROUP BY agent_id, command_type, command_status;
|
||||
|
||||
5. Reconcile newly cancelled rows to agent.command.rejected records. Check command ID,
|
||||
Agent ID, prior state, immutable rejection reason, source and operator attribution.
|
||||
Current pending status does not establish that a command was never delivered.
|
||||
Current claimed status, any retained claimed_at, completion/result/error evidence,
|
||||
linked legacy result or non-creation audit history, an unlinked legacy result
|
||||
for that Agent, or an Agent result submission
|
||||
requires execution_outcome=unknown. Missing or ambiguous evidence is unknown.
|
||||
Only a pending row with a complete pre-cancellation check finding no delivery
|
||||
evidence is classified not_dispatched by the queue. Historical claim evidence
|
||||
survives cancellation; an explicitly repaired/requeued pending row is still
|
||||
unknown. No ordinary production requeue mechanism is introduced by Stage A.
|
||||
Neither classification proves absence of out-of-band historical physical writes.
|
||||
Repeated retirement must add no duplicate audit and must not rewrite history.
|
||||
6. Schedule cancel/delete uses the stored schedule's Agent ID and the same rejection
|
||||
audit policy; delete retains the schedule. This is NOT a fleet drain: orphaned or
|
||||
foreign-Agent rows are handled by the per-Agent cutover sweep above. Never delete
|
||||
queue/audit rows, reset claimed commands to pending, or rewrite terminal results.
|
||||
Existing pre-remediation cancelled rows missing an audit must be preserved and
|
||||
documented as a historical evidence gap. Do not invent a prior-state audit.
|
||||
|
||||
Exit gate: the global live-unsupported query returns no rows, locks are accounted
|
||||
for, transition counts/audits reconcile, and no old producer can replenish work.
|
||||
This gate does not revoke already delivered payloads; gates 1, 2 and 5 remain mandatory.
|
||||
|
||||
Gate 5: upgrade Windows code and verify installed bytes while stopped
|
||||
|
||||
1. Use a verified artifact prepared from the exact independently reviewed Stage A
|
||||
boundary. Record its digest and a trusted expected SHA-256 manifest outside the
|
||||
target installation. VERSION is unchanged at 1.0.0-alpha and is not upgrade proof.
|
||||
2. Include all managed runtime scripts, especially MinerOpsAgentService.ps1,
|
||||
MinerOpsControl.ps1, EnrichOneMiner.ps1, MinerOpsApi3EnrichmentPoller.ps1, the pool
|
||||
pollers, task/GUI/tray scripts, and VERSION. Check the package's installer and
|
||||
enrollment/credential-upgrade entry points as well. Include the manifest's origin
|
||||
and review approval; hashes computed only from a suspect host are not a trust root.
|
||||
3. Confirm the main task was positively stopped BEFORE invoking the existing
|
||||
installer. It can auto-restart a task that it found running. Keep the external
|
||||
network fence and restart hold throughout installation and failure recovery.
|
||||
Do not pass filesystem trust-reset switches unless their existing reviewed
|
||||
recovery workflow specifically calls for them. Preserve config/state/logs/backups.
|
||||
4. Compare installed Get-FileHash -Algorithm SHA256 values to the trusted manifest.
|
||||
Review task actions for the expected executable, canonical script path, arguments,
|
||||
principal, and original/reviewed definition; check all aliases/duplicate tasks.
|
||||
Verify ACLs, signer/source provenance where used, absence of old running processes,
|
||||
and the task state consistent with the maintenance hold. Do not overwrite an
|
||||
established task definition merely to force a convenient state.
|
||||
5. Follow the existing credential transition/enrollment recovery workflow only if
|
||||
required. Hub credentials and local miner credentials are different trust domains;
|
||||
rotating a Hub secret cannot recall a cached physical command. Keep credentials
|
||||
local and protected. Never restore an old credential/config to make a test pass.
|
||||
|
||||
Exit gate: installed bytes and process/task state match the approved ledger for
|
||||
every Agent proposed for release. An inaccessible Agent, hash mismatch, duplicate
|
||||
task, unexpected process, or trust-reset blocker remains fenced and stopped.
|
||||
|
||||
Gate 6: native Windows validation and fresh process startup
|
||||
|
||||
Before production release, run on a network-isolated native Windows PowerShell 5.1
|
||||
host with simulated miners only. Linux PowerShell 7 tests are not evidence of SYSTEM
|
||||
task lifecycle correctness. Validate these points and retain results:
|
||||
- Actual SYSTEM/Scheduled Task stop, child cleanup, restart hold and reboot behavior.
|
||||
- Installed package hash verification and fresh module loading after upgrade.
|
||||
- Legacy/malformed/unsupported cached commands rejected with zero physical writes.
|
||||
- Diagnostic write/parameter overrides rejected before any device connection;
|
||||
fixed telemetry/discovery/API3 read selections remain compatible with simulators.
|
||||
- Enrollment, credential-transition interruption/recovery, and running/stopped intent
|
||||
retain existing protections, without credentials appearing in normal output.
|
||||
- A full network fence prevents an old cached command from reaching a live simulator;
|
||||
a separate unfenced positive control proves the simulator and old-executor setup.
|
||||
|
||||
1. Start only a fresh verified Agent process, initially still fenced, after all
|
||||
preceding gates are satisfied. Verify executable/script paths, start time after
|
||||
publication, loaded source provenance, per-Agent authentication, Zone assignment,
|
||||
and unavailable control advertisement. The advertisement alone is not proof.
|
||||
2. Recheck the unsupported queue and audit reconciliation after startup. Do not
|
||||
restore read connectivity for a host with any unverified old executor.
|
||||
3. Release the fence only for individually verified upgraded hosts under the approved
|
||||
network change. Re-enable only reviewed telemetry schedules. Physical control
|
||||
stays disabled on Hub AND Agent. Retain independent monitor/rollback evidence.
|
||||
|
||||
Exit gate: every released Agent has complete positive verification. Fleet-wide
|
||||
containment cannot be claimed while unverified devices have an unfenced route.
|
||||
|
||||
Gate 7: returning offline Agents and rollback/recovery
|
||||
|
||||
Maintain a default quarantine for every offline, unknown, historical or unverified
|
||||
Agent when it reconnects. A timestamp, IP address, same VERSION, successful Hub
|
||||
authentication, or control_available=false claim cannot release it. Repeat stop,
|
||||
hash/task verification, upgrade, fresh-start and relevant gates before restoring
|
||||
miner connectivity. Keep retired Agent credentials retired; do not reopen legacy
|
||||
shared-secret fallback for convenience.
|
||||
|
||||
If ANY gate fails:
|
||||
- Keep or reinstate physical network fencing immediately. Stop/restart-hold Agent
|
||||
executors and prevent old Hub producers from running. Preserve evidence and notify
|
||||
the change owner; do not automatically resume the previous running state.
|
||||
- Database/audit failure: leave rolled-back rows as evidence, resolve the error,
|
||||
retry audited retirement, and re-run the global zero-live check. Never repair
|
||||
audit integrity by deleting history or fabricating execution outcomes.
|
||||
- Installer/recovery failure: preserve trust blockers, recovery tokens and protected
|
||||
backups according to existing recovery instructions. Verify state before retry.
|
||||
- If restoring old Hub or Agent binaries is unavoidable, those binaries stay stopped
|
||||
and all affected miner routes stay fenced. An operational rollback does not restore
|
||||
security containment. Do not reconnect until the reviewed Stage A bytes are again
|
||||
installed and every failed gate is repeated successfully.
|
||||
- If verification cannot be completed within the change window, leave the affected
|
||||
host quarantined with a named owner. Partial success is not authority to un-fence it.
|
||||
|
||||
Disposable simulation available for independent review
|
||||
|
||||
agents/windows/tests/Test-MinerOpsOldAgentFence.ps1 loads AST functions from a
|
||||
separate git archive of 2d89e80 and the audit's cached suspend payload. It requires
|
||||
MINEROPS_STAGE_A_SIMULATION=1 and an explicitly provided simulated target. The
|
||||
remediation artifact directory contains simulate_fence.py and simulated_miner.py
|
||||
which create a temporary Docker INTERNAL network and fake API3 miner, then run the
|
||||
old executor first with --network none (zero simulated writes) and separately on
|
||||
the internal network (one simulated service-stop write). No host ports are exposed.
|
||||
Only those created resources are removed. This demonstrates the coarse fence's
|
||||
effect, not the adequacy of any production firewall/NAC design. Native Windows
|
||||
evidence and actual fleet/network verification remain separate release gates.
|
||||
Executable
+40
@@ -0,0 +1,40 @@
|
||||
#!/usr/bin/env bash
|
||||
# Run ONLY from the isolated review candidate. No .env files or published DB ports.
|
||||
set -euo pipefail
|
||||
review_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
: "${MINEROPS_REVIEW_BACKEND_IMAGE:?Set an existing Python backend image with requirements installed}"
|
||||
docker image inspect "$MINEROPS_REVIEW_BACKEND_IMAGE" >/dev/null
|
||||
review_id="minerops-discovery-review-$$-$RANDOM"
|
||||
review_network="$review_id-network"
|
||||
review_db="$review_id-postgres"
|
||||
cleanup() {
|
||||
docker rm -f "$review_db" >/dev/null 2>&1 || true
|
||||
docker network rm "$review_network" >/dev/null 2>&1 || true
|
||||
}
|
||||
trap cleanup EXIT
|
||||
docker network create --internal "$review_network" >/dev/null
|
||||
docker run -d --name "$review_db" --network "$review_network" --network-alias postgres \
|
||||
--tmpfs /var/lib/postgresql/data \
|
||||
-e POSTGRES_DB=discovery_review -e POSTGRES_USER=reviewer \
|
||||
-e POSTGRES_PASSWORD=disposable-review-only postgres:16-alpine >/dev/null
|
||||
review_ready=0
|
||||
for _ in $(seq 1 40); do
|
||||
if docker exec "$review_db" pg_isready -U reviewer -d discovery_review >/dev/null 2>&1; then review_ready=1; break; fi
|
||||
sleep 0.5
|
||||
done
|
||||
[[ "$review_ready" == 1 ]]
|
||||
run_backend() {
|
||||
docker run --rm --network "$review_network" -v "$review_root:/review:ro" \
|
||||
-w /review/backend -e PYTHONDONTWRITEBYTECODE=1 -e PYTHONPATH=/review/backend \
|
||||
-e POSTGRES_HOST=postgres -e POSTGRES_DB=discovery_review -e POSTGRES_USER=reviewer \
|
||||
-e POSTGRES_PASSWORD=disposable-review-only -e JWT_SECRET_KEY=disposable-review-only \
|
||||
-e MINEROPS_DISCOVERY_DISPOSABLE=1 -e MINEROPS_HUB_MODE=true \
|
||||
-e SCHEDULED_SCAN_ENABLED=false -e MINEROPS_WINDOWS_AGENT_SOURCE=/review/agents/windows \
|
||||
"$MINEROPS_REVIEW_BACKEND_IMAGE" "$@"
|
||||
}
|
||||
run_backend alembic upgrade head
|
||||
run_backend python -m unittest -v tests.test_discovery_security_postgres \
|
||||
tests.test_agent_discovery_isolation_postgres \
|
||||
tests.test_legacy_inventory_security_postgres
|
||||
# Browser and PowerShell suites use their own intercepted/loopback transports.
|
||||
# See docs/operations/BLOCKER_1_DISCOVERY_REVIEW.txt for exact commands and baseline exceptions.
|
||||
@@ -16,7 +16,7 @@ cleanup() {
|
||||
}
|
||||
trap cleanup EXIT
|
||||
[[ -f .env.dev ]] || { echo "Missing .env.dev" >&2; exit 1; }
|
||||
for source_file in backend/tests/test_agent_auth.py backend/tests/test_miner_control.py backend/tests/test_agent_commands.py backend/tests/test_miner_control_postgres.py; do
|
||||
for source_file in backend/tests/test_agent_auth.py backend/tests/test_miner_control.py backend/tests/test_agent_commands.py backend/tests/test_miner_control_postgres.py backend/tests/test_legacy_control_retirement.py; do
|
||||
permissions="$(stat -c '%a' "${source_file}")"
|
||||
(( (10#${permissions} % 10) & 4 )) || {
|
||||
echo "${source_file} is not readable by the non-root test container" >&2
|
||||
@@ -47,6 +47,7 @@ done
|
||||
tests.test_agent_auth \
|
||||
tests.test_miner_control \
|
||||
tests.test_agent_commands \
|
||||
tests.test_legacy_control_retirement \
|
||||
tests.test_miner_control_postgres
|
||||
|
||||
echo "Miner Control Foundation focused tests passed."
|
||||
Executable
+28
@@ -0,0 +1,28 @@
|
||||
#!/usr/bin/env bash
|
||||
# Run inside an isolated test environment with a freshly migrated disposable DB.
|
||||
# No service creation, cleanup, miner commands, or production defaults.
|
||||
set -euo pipefail
|
||||
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
cd "${repo_root}/backend"
|
||||
[[ "${MINEROPS_STAGE_A_DISPOSABLE:-}" == 1 ]] || {
|
||||
echo "Require MINEROPS_STAGE_A_DISPOSABLE=1 and an isolated test database." >&2
|
||||
exit 1
|
||||
}
|
||||
export PYTHONDONTWRITEBYTECODE=1
|
||||
export MINEROPS_ENABLE_REAL_REBOOT=false
|
||||
export MINEROPS_ENABLE_REAL_POWER_CONTROL=false
|
||||
python -c 'from app.config import settings; assert settings.postgres_db.startswith("minerops_stage_a_"), "Refusing non-Stage-A database"'
|
||||
python -m unittest -v \
|
||||
tests.test_legacy_control_retirement \
|
||||
tests.test_stage_a_containment \
|
||||
tests.test_stage_a_containment_postgres \
|
||||
tests.test_stage_a_delivery_evidence \
|
||||
tests.test_stage_a_delivery_evidence_postgres \
|
||||
tests.test_stage_a_schedule_cancellation_postgres \
|
||||
tests.test_agent_zones_postgres \
|
||||
tests.test_miner_control \
|
||||
tests.test_agent_commands.AgentCommandServiceTests \
|
||||
tests.test_agent_commands.AgentCommandRouterTests
|
||||
# Separately run agents/windows/tests/Test-MinerOpsControl.ps1 in a network-free
|
||||
# PowerShell environment, then Test-MinerOpsDiagnosticTransport.ps1. Both load
|
||||
# functions only. Diagnostic compatibility listeners bind loopback exclusively.
|
||||
Reference in new issue
Block a user