From 8b0d32cd6b64c9b1100485804fdd78b9d9eb0c56 Mon Sep 17 00:00:00 2001 From: JC Date: Sat, 17 Oct 2026 00:45:04 +0000 Subject: [PATCH] security(discovery): enforce Zone isolation and retire unsafe legacy writers Bind Agent observations, review, enrichment and telemetry to current Zone and inventory authority. Preserve immutable MAC ownership and historical snapshots with stale-result rejection and transactional security audits. Retire unbound direct/range/scheduled discovery and one/bulk legacy summary writes in both modes. Require reconciled exact inventory identity or explicit Zone-scoped IP selection for CSV metadata imports. Independent Blocker #1 application-security re-audit: PASS on tree cf863aadc9019aefcdec3ba1dd3880619f4f6561. Evidence: 141 focused PostgreSQL/HTTP tests passed; Chromium and PowerShell checks passed. Two documented baseline source assertions remain unchanged. Production deployment is not approved. --- agents/windows/EnrichOneMiner.ps1 | 16 +- agents/windows/MinerOpsAgentService.ps1 | 25 +- .../windows/MinerOpsApi3EnrichmentPoller.ps1 | 23 +- agents/windows/MinerOpsPoolPoller.ps1 | 21 +- .../windows/tests/Test-MinerOpsDiscovery.ps1 | 55 + backend/app/agent_commands/repository.py | 22 + backend/app/agent_zones/models.py | 5 + backend/app/agent_zones/repository.py | 28 +- backend/app/agent_zones/service.py | 27 + backend/app/discovery/__init__.py | 1 + backend/app/discovery/csv_import.py | 96 + backend/app/discovery/http.py | 42 + backend/app/discovery/repository.py | 1260 +++++ backend/app/identity.py | 3 + backend/app/main.py | 4141 +---------------- backend/tests/browser/discovery-security.cjs | 49 + .../tests/browser/export_discovery_page.py | 20 + backend/tests/test_agent_commands.py | 103 +- ...test_agent_discovery_isolation_postgres.py | 39 + backend/tests/test_agent_zones.py | 81 +- backend/tests/test_agent_zones_postgres.py | 6 +- .../tests/test_discovery_security_postgres.py | 1211 +++++ ...test_legacy_inventory_security_postgres.py | 338 ++ .../test_stage_a_containment_postgres.py | 19 +- .../operations/BLOCKER_1_DISCOVERY_REVIEW.txt | 219 + scripts/test-discovery-security.sh | 40 + 26 files changed, 3798 insertions(+), 4092 deletions(-) create mode 100644 agents/windows/tests/Test-MinerOpsDiscovery.ps1 create mode 100644 backend/app/discovery/__init__.py create mode 100644 backend/app/discovery/csv_import.py create mode 100644 backend/app/discovery/http.py create mode 100644 backend/app/discovery/repository.py create mode 100644 backend/tests/browser/discovery-security.cjs create mode 100644 backend/tests/browser/export_discovery_page.py create mode 100644 backend/tests/test_agent_discovery_isolation_postgres.py create mode 100644 backend/tests/test_discovery_security_postgres.py create mode 100644 backend/tests/test_legacy_inventory_security_postgres.py create mode 100644 docs/operations/BLOCKER_1_DISCOVERY_REVIEW.txt create mode 100755 scripts/test-discovery-security.sh diff --git a/agents/windows/EnrichOneMiner.ps1 b/agents/windows/EnrichOneMiner.ps1 index 2c0d75c..c852218 100644 --- a/agents/windows/EnrichOneMiner.ps1 +++ b/agents/windows/EnrichOneMiner.ps1 @@ -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" @@ -164,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", @@ -219,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 diff --git a/agents/windows/MinerOpsAgentService.ps1 b/agents/windows/MinerOpsAgentService.ps1 index 01f10fe..0fb0a91 100644 --- a/agents/windows/MinerOpsAgentService.ps1 +++ b/agents/windows/MinerOpsAgentService.ps1 @@ -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 @@ -1015,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" diff --git a/agents/windows/MinerOpsApi3EnrichmentPoller.ps1 b/agents/windows/MinerOpsApi3EnrichmentPoller.ps1 index 5b112a0..cfb90df 100644 --- a/agents/windows/MinerOpsApi3EnrichmentPoller.ps1 +++ b/agents/windows/MinerOpsApi3EnrichmentPoller.ps1 @@ -224,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 @@ -257,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 @@ -275,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" diff --git a/agents/windows/MinerOpsPoolPoller.ps1 b/agents/windows/MinerOpsPoolPoller.ps1 index 3773e8f..a70fad4 100644 --- a/agents/windows/MinerOpsPoolPoller.ps1 +++ b/agents/windows/MinerOpsPoolPoller.ps1 @@ -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 diff --git a/agents/windows/tests/Test-MinerOpsDiscovery.ps1 b/agents/windows/tests/Test-MinerOpsDiscovery.ps1 new file mode 100644 index 0000000..166c3b9 --- /dev/null +++ b/agents/windows/tests/Test-MinerOpsDiscovery.ps1 @@ -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." diff --git a/backend/app/agent_commands/repository.py b/backend/app/agent_commands/repository.py index bb53511..a784a67 100644 --- a/backend/app/agent_commands/repository.py +++ b/backend/app/agent_commands/repository.py @@ -10,6 +10,8 @@ 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] @@ -120,9 +122,29 @@ class AgentCommandRepository: connection = self._connection_factory() try: with connection.cursor(row_factory=dict_row) as cur: + 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: diff --git a/backend/app/agent_zones/models.py b/backend/app/agent_zones/models.py index 987774c..51a2670 100644 --- a/backend/app/agent_zones/models.py +++ b/backend/app/agent_zones/models.py @@ -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 diff --git a/backend/app/agent_zones/repository.py b/backend/app/agent_zones/repository.py index 8c3680a..8ead4c4 100644 --- a/backend/app/agent_zones/repository.py +++ b/backend/app/agent_zones/repository.py @@ -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: diff --git a/backend/app/agent_zones/service.py b/backend/app/agent_zones/service.py index 5b1685a..9d5e0d0 100644 --- a/backend/app/agent_zones/service.py +++ b/backend/app/agent_zones/service.py @@ -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: diff --git a/backend/app/discovery/__init__.py b/backend/app/discovery/__init__.py new file mode 100644 index 0000000..52152b4 --- /dev/null +++ b/backend/app/discovery/__init__.py @@ -0,0 +1 @@ +"""Agent discovery isolation; physical miner writes are never authorized here.""" diff --git a/backend/app/discovery/csv_import.py b/backend/app/discovery/csv_import.py new file mode 100644 index 0000000..2496093 --- /dev/null +++ b/backend/app/discovery/csv_import.py @@ -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 diff --git a/backend/app/discovery/http.py b/backend/app/discovery/http.py new file mode 100644 index 0000000..b483900 --- /dev/null +++ b/backend/app/discovery/http.py @@ -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") diff --git a/backend/app/discovery/repository.py b/backend/app/discovery/repository.py new file mode 100644 index 0000000..5342254 --- /dev/null +++ b/backend/app/discovery/repository.py @@ -0,0 +1,1260 @@ +"""Transactional discovery authority and append-only evidence, using existing schema. + +All writers acquire bounded table locks in the same order. The miners lock also +serializes INSERT phantoms with legacy writers which do not take our advisory +lock. This deliberately favors correctness over fleet-scale write concurrency. +No miner I/O occurs here. No discovered address authorizes a physical command. +""" + +from contextlib import contextmanager +from datetime import datetime, timezone, timedelta +from hashlib import sha256 +import ipaddress +import json +import re +import math +from uuid import UUID, uuid4, uuid5, NAMESPACE_URL + +import jwt +import psycopg +from psycopg.rows import dict_row +from psycopg.types.json import Jsonb +from fastapi import HTTPException + +from app.config import settings +from app.agent_zones.repository import default_connection_factory + +TTL = timedelta(minutes=15) +MAX_BATCH = 128 + + +def reject(reason, code=409): + raise HTTPException(status_code=code, detail=reason) + + +def object_body(body): + if not isinstance(body, dict): + reject("discovery_object_required", 422) + return body + + +def address(value): + try: + text = str(value) + if "%" in text or "/" in text: + raise ValueError() + return str(ipaddress.ip_address(text)) + except (ValueError, TypeError): + reject("discovery_invalid_ip", 422) + + +def mac(value): + if value in (None, ""): + return None + if not isinstance(value, str) or not re.fullmatch( + r"(?:[0-9a-fA-F]{12}|(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}|(?:[0-9a-fA-F]{2}-){5}[0-9a-fA-F]{2})", + value, + ): + reject("discovery_invalid_mac", 422) + key = value.replace(":", "").replace("-", "").lower() + if key in {"000000000000", "ffffffffffff"} or int(key[:2], 16) & 1: + reject("discovery_invalid_mac", 422) + return ":".join(key[i : i + 2] for i in range(0, 12, 2)).upper() + + +def digest(value): + return sha256( + json.dumps(value, sort_keys=True, default=str, separators=(",", ":")).encode() + ).hexdigest() + + +def text(value, limit=255): + if value is None: + return None + if not isinstance(value, str) or len(value) > limit: + reject("discovery_invalid_text", 422) + return value.strip() or None + + +def number(value, integer=False): + if value is None or value == "": + return None + try: + if isinstance(value, bool): + raise ValueError() + parsed = float(value) + if not math.isfinite(parsed): + raise ValueError() + return int(parsed) if integer else parsed + except (ValueError, TypeError, OverflowError): + reject("discovery_invalid_measurement", 422) + + +def insert(cur, table, values): + # Identifiers originate exclusively from fixed mappings in this module. + cur.execute( + "INSERT INTO " + + table + + "(" + + ",".join(values) + + ") VALUES (" + + ",".join(["%s"] * len(values)) + + ")", + tuple(values.values()), + ) + + +def status_snapshot(cur, row, raw, api3=False): + raw = object_body(raw or {}) + if api3: + msg = object_body(object_body(raw.get("msg") or {}).get("summary") or {}) + names = { + "hash-average": "mhs_av", + "power-realtime": "power_w", + "power-rate": "power_rate_w_per_th", + "chip-temp-avg": "chip_temp_avg_c", + "chip-temp-max": "chip_temp_max_c", + "environment-temperature": "env_temp_c", + } + else: + msg = object_body(raw.get("Msg") or raw.get("msg") or {}) + names = { + "MHS av": "mhs_av", + "MHS 1m": "mhs_1m", + "MHS 15m": "mhs_15m", + "HS RT": "hs_rt", + "Power": "power_w", + "Power Rate": "power_rate_w_per_th", + "Env Temp": "env_temp_c", + "Chip Temp Min": "chip_temp_min_c", + "Chip Temp Max": "chip_temp_max_c", + "Chip Temp Avg": "chip_temp_avg_c", + } + values = { + "miner_id": row["id"], + "observed_ip": row["current_ip"], + "raw_summary_json": Jsonb(raw), + } + for key, column in names.items(): + if key in msg: + values[column] = number(msg[key]) + if api3 and values.get("mhs_av") is not None: + values["mhs_av"] *= 1000000 + for key, column in ( + { + "fan-speed-in": "fan_speed_in_rpm", + "fan-speed-out": "fan_speed_out_rpm", + "bootup-time": "uptime_seconds", + } + if api3 + else { + "Fan Speed In": "fan_speed_in_rpm", + "Fan Speed Out": "fan_speed_out_rpm", + "Uptime": "uptime_seconds", + } + ).items(): + if key in msg: + values[column] = number(msg[key], True) + insert(cur, "miner_status_snapshots", values) + + +def pool_snapshot(cur, row, pool, agent_id): + object_body(pool) + values = { + "miner_id": row["id"], + "ip_address": row["current_ip"], + "mac_address": row["current_mac"], + "raw_pool_json": Jsonb(pool), + "agent_id": agent_id, + } + for keys, column in [ + (("URL", "url"), "url"), + (("User", "account", "user"), "worker_user"), + (("Status", "status"), "pool_status"), + ]: + v = next((pool[k] for k in keys if k in pool), None) + if v is not None: + values[column] = text(v, 2048) + for keys, column in [ + (("POOL", "id"), "pool_index"), + (("Accepted", "accepted"), "accepted"), + (("Rejected", "rejected"), "rejected"), + ]: + v = next((pool[k] for k in keys if k in pool), None) + if v is not None: + values[column] = number(v, True) + insert(cur, "miner_pool_snapshots", values) + + +class DiscoveryRepository: + def __init__(self, connection_factory=default_connection_factory): + self.connection_factory = connection_factory + + @contextmanager + def transaction(self): + with self.connection_factory() as conn: + with conn.cursor(row_factory=dict_row) as cur: + self.lock(cur) + yield cur + + @staticmethod + def lock(cur): + cur.execute("SET LOCAL lock_timeout = '5s'") + cur.execute("SET LOCAL statement_timeout = '15s'") + cur.execute("SELECT pg_advisory_xact_lock(743190031)") + cur.execute( + "LOCK TABLE minerops_site_agents, agent_zones, agent_zone_network_ranges IN SHARE MODE" + ) + cur.execute("LOCK TABLE miners IN SHARE ROW EXCLUSIVE MODE") + + @staticmethod + def audit(cur, action, entity, value, actor=None, old=None): + cur.execute( + """INSERT INTO audit_logs + (actor_user_id, action, entity_type, entity_id, old_value_json, new_value_json) + VALUES (%s, %s, 'discovery', %s, %s, %s)""", + ( + actor, + action, + entity, + Jsonb(old) if old is not None else None, + Jsonb(value), + ), + ) + + @staticmethod + def context(cur, agent_id): + cur.execute( + """SELECT a.agent_id, a.site_id, a.agent_zone_id, + a.credential_created_at, a.credential_revoked_at, + z.updated_at AS zone_version, z.is_enabled, s.name AS site_name + FROM minerops_site_agents a JOIN agent_zones z ON z.id=a.agent_zone_id + AND z.site_id=a.site_id JOIN sites s ON s.id=a.site_id + WHERE a.agent_id=%s FOR SHARE OF a,z""", + (agent_id,), + ) + row = cur.fetchone() + if not row or not row["is_enabled"] or row["credential_revoked_at"] is not None: + reject("discovery_routing_unavailable") + cur.execute( + "SELECT network::text FROM agent_zone_network_ranges WHERE agent_zone_id=%s ORDER BY network,id FOR SHARE", + (row["agent_zone_id"],), + ) + row["networks"] = [v["network"] for v in cur.fetchall()] + if not row["networks"]: + reject("discovery_network_unassigned") + row["authority"] = digest( + { + k: row[k] + for k in ( + "agent_id", + "site_id", + "agent_zone_id", + "credential_created_at", + "zone_version", + "networks", + ) + } + ) + return row + + @staticmethod + def in_network(ctx, ip): + ip = address(ip) + if not any( + ipaddress.ip_address(ip) in ipaddress.ip_network(n) for n in ctx["networks"] + ): + reject("discovery_address_outside_zone") + + @staticmethod + def identity(body, agent_id): + object_body(body) + if body.get("agent_id") not in (None, "", agent_id): + reject("discovery_agent_identity_conflict", 403) + + @staticmethod + def fingerprint(row): + return ( + digest( + { + k: row.get(k) + for k in ( + "id", + "site_id", + "agent_zone_id", + "current_mac", + "current_ip", + "updated_at", + ) + } + ) + if row + else None + ) + + def match(self, cur, ctx, item): + """All supplied identifiers must agree. No global IP search or ownership claim.""" + if ( + item.get("ip_address") + and item.get("ip") + and address(item["ip_address"]) != address(item["ip"]) + ): + reject("discovery_identity_conflict") + if ( + item.get("mac_address") + and item.get("mac") + and mac(item["mac_address"]) != mac(item["mac"]) + ): + reject("discovery_identity_conflict") + ip = address(item.get("ip_address") or item.get("ip")) + self.in_network(ctx, ip) + supplied_mac = mac(item.get("mac_address") or item.get("mac")) + candidates = [] + if item.get("miner_id"): + try: + mid = UUID(str(item["miner_id"])) + except ValueError: + reject("discovery_invalid_miner_id", 422) + cur.execute( + "SELECT * FROM miners WHERE id=%s AND agent_zone_id=%s FOR UPDATE", + (mid, ctx["agent_zone_id"]), + ) + row = cur.fetchone() + if not row: + reject("discovery_identity_conflict") + candidates.append(row) + if supplied_mac: + key = supplied_mac.replace(":", "").lower() + cur.execute( + """SELECT m.* FROM miner_mac_ownerships o JOIN miners m ON m.id=o.miner_id + WHERE o.normalized_mac=%s FOR UPDATE OF m""", + (key,), + ) + row = cur.fetchone() + if row: + if ( + row["agent_zone_id"] != ctx["agent_zone_id"] + or mac(row["current_mac"]) != supplied_mac + ): + reject("discovery_identity_conflict") + candidates.append(row) + else: + cur.execute( + """SELECT * FROM miners WHERE agent_zone_id=%s + AND regexp_replace(lower(current_mac),'[^0-9a-f]','','g')=%s FOR UPDATE""", + (ctx["agent_zone_id"], key), + ) + rows = cur.fetchall() + if len(rows) > 1: + reject("discovery_identity_ambiguous") + candidates.extend(rows) + serial = text(item.get("miner_sn")) + if serial: + cur.execute( + "SELECT * FROM miners WHERE miner_sn=%s ORDER BY id LIMIT 2 FOR UPDATE", + (serial,), + ) + serial_rows = cur.fetchall() + if len(serial_rows) > 1 or any( + r["agent_zone_id"] != ctx["agent_zone_id"] for r in serial_rows + ): + reject("discovery_identity_conflict") + candidates.extend(serial_rows) + cur.execute( + "SELECT * FROM miners WHERE agent_zone_id=%s AND host(current_ip)=%s ORDER BY id LIMIT 2 FOR UPDATE", + (ctx["agent_zone_id"], ip), + ) + rows = cur.fetchall() + if len(rows) > 1: + reject("discovery_identity_ambiguous") + candidates.extend(rows) + if len({r["id"] for r in candidates}) > 1: + reject("discovery_identity_conflict") + row = candidates[0] if candidates else None + if row and ( + str(row["current_ip"]) != ip + or (supplied_mac and mac(row["current_mac"]) != supplied_mac) + or (serial and row["miner_sn"] != serial) + ): + reject("discovery_identity_conflict") + return row + + @staticmethod + def observation_version(row): + return row["updated_at"].isoformat() + + def observed(self, cur, ctx, item, parent=None): + ip = address(item.get("ip_address")) + self.in_network(ctx, ip) + canonical_mac = mac(item.get("mac_address")) + if item.get("discovered_at"): + try: + reported = datetime.fromisoformat( + item["discovered_at"].replace("Z", "+00:00") + ) + if ( + reported.tzinfo is None + or not -timedelta(minutes=5) + <= datetime.now(timezone.utc) - reported + <= TTL + ): + raise ValueError() + except (ValueError, AttributeError, TypeError): + reject("discovery_stale_report", 422) + # Input is kept verbatim; authoritative provenance exists only in audit_logs. + oid = str( + uuid5( + NAMESPACE_URL, + "minerops-discovery:" + ctx["authority"] + ":" + digest(item), + ) + ) + cur.execute( + "SELECT * FROM minerops_agent_discoveries WHERE id=%s FOR UPDATE", (oid,) + ) + existing = cur.fetchone() + if existing: + return existing + status = "new" + reason = None + matched = None + try: + matched = self.match( + cur, + ctx, + { + **item, + "ip_address": ip, + "mac_address": canonical_mac, + "miner_sn": item.get("miner_sn"), + }, + ) + except HTTPException as exc: + if exc.status_code != 409: + raise + status = "zone_conflict" + reason = exc.detail + if matched: + status = "known" + cur.execute( + """INSERT INTO minerops_agent_discoveries + (id,agent_id,agent_zone_id,site_name,room_name,ip_address,mac_address,model,hostname,miner_sn, + raw_payload,discovery_status,matched_miner_id) + VALUES (%s,%s,%s,%s,%s,%s::inet,%s,%s,%s,%s,%s,%s,%s) RETURNING *""", + ( + oid, + ctx["agent_id"], + ctx["agent_zone_id"], + ctx["site_name"], + text(item.get("room_name")), + ip, + canonical_mac, + text(item.get("model")), + text(item.get("hostname")), + text(item.get("miner_sn")), + Jsonb(item), + status, + matched["id"] if matched else None, + ), + ) + row = cur.fetchone() + self.audit( + cur, + "discovery.observed", + oid, + { + "agent_id": ctx["agent_id"], + "authority": ctx["authority"], + "payload_hash": digest(item), + "miner_fingerprint": self.fingerprint(matched), + "reason": reason, + "parent_observation_id": parent, + "version": self.observation_version(row), + }, + ) + return row + + def ingest(self, agent_id, body): + self.identity(body, agent_id) + items = body.get("discoveries") + if not isinstance(items, list) or len(items) > MAX_BATCH: + reject("discovery_invalid_batch", 422) + if len({address(object_body(i).get("ip_address")) for i in items}) != len( + items + ): + reject("discovery_duplicate_batch_address", 422) + with self.transaction() as cur: + ctx = self.context(cur, agent_id) + cur.execute( + """SELECT count(*) FROM audit_logs WHERE action='discovery.observed' + AND new_value_json->>'agent_id'=%s AND created_at>clock_timestamp()-interval '1 minute'""", + (agent_id,), + ) + if cur.fetchone()["count"] + len(items) > 1024: + reject("discovery_rate_limit", 429) + rows = [self.observed(cur, ctx, object_body(i)) for i in items] + return { + "ok": True, + "received": len(items), + "inserted": len(rows), + "matched_existing": sum(r["discovery_status"] == "known" for r in rows), + "staged_new": sum(r["discovery_status"] == "new" for r in rows), + "conflicts": sum( + r["discovery_status"] == "zone_conflict" for r in rows + ), + "agent_zone_id": str(ctx["agent_zone_id"]), + "routing_reason": None, + "errors": [], + "discoveries": [self.present(r) for r in rows], + } + + @staticmethod + def present(row): + return { + **{ + k: ( + str(v) + if isinstance( + v, + (UUID, datetime, ipaddress.IPv4Address, ipaddress.IPv6Address), + ) + else v + ) + for k, v in row.items() + if k != "raw_payload" + }, + "version": row["updated_at"].isoformat(), + } + + def listing(self): + with self.connection_factory() as conn, conn.cursor( + row_factory=dict_row + ) as cur: + cur.execute( + "SELECT * FROM minerops_agent_discoveries ORDER BY created_at DESC,id LIMIT 100" + ) + return { + "ok": True, + "discoveries": [self.present(r) for r in cur.fetchall()], + } + + def observation(self, cur, oid, version): + try: + UUID(str(oid)) + except (ValueError, TypeError): + reject("discovery_id_required", 422) + cur.execute( + "SELECT * FROM minerops_agent_discoveries WHERE id=%s FOR UPDATE", + (str(oid),), + ) + row = cur.fetchone() + if not row: + reject("discovery_not_found", 404) + ctx = self.context(cur, row["agent_id"]) + if row["agent_zone_id"] != ctx["agent_zone_id"]: + reject("discovery_stale_authority") + cur.execute( + "SELECT new_value_json,created_at FROM audit_logs WHERE entity_type='discovery' AND entity_id=%s AND action='discovery.observed' ORDER BY created_at LIMIT 2", + (str(oid),), + ) + proofs = cur.fetchall() + if len(proofs) != 1: + reject("discovery_unverified_history") + proof = proofs[0] + if proof["new_value_json"].get("authority") != ctx["authority"] or proof[ + "new_value_json" + ].get("payload_hash") != digest(row["raw_payload"]): + reject("discovery_stale_authority") + if self.observation_version(row) != proof["new_value_json"].get("version"): + cur.execute( + "SELECT new_value_json FROM audit_logs WHERE entity_id=%s AND action='discovery.reviewed' ORDER BY created_at DESC LIMIT 1", + (str(oid),), + ) + decision = cur.fetchone() + if not decision: + reject("discovery_unverified_history") + saved = decision["new_value_json"] + response = saved.get("response", {}) + expected_state = {"ignore": "ignored", "non_miner": "non_miner"}.get( + saved.get("action") + ) + if expected_state is None: + expected_state = ( + "known" + if proof["new_value_json"].get("miner_fingerprint") + else "added" + ) + if ( + response.get("version") != self.observation_version(row) + or response.get("miner_id") + != (str(row["matched_miner_id"]) if row["matched_miner_id"] else None) + or row["discovery_status"] != expected_state + ): + reject("discovery_unverified_history") + self.in_network(ctx, str(row["ip_address"])) + if datetime.now(timezone.utc) - row["created_at"] > TTL: + reject("discovery_stale_observation") + cur.execute( + """SELECT 1 FROM minerops_agent_discoveries WHERE agent_id=%s AND agent_zone_id=%s + AND ip_address=%s AND id<>%s AND created_at>=%s LIMIT 1""", + ( + row["agent_id"], + row["agent_zone_id"], + row["ip_address"], + row["id"], + row["created_at"], + ), + ) + if cur.fetchone(): + reject("discovery_superseded_observation") + return row, ctx, proof["new_value_json"] + + def review(self, body, user): + object_body(body) + oid = body.get("discovery_id") + version = body.get("version") + action = body.get("action") + if not isinstance(action, str) or action not in { + "add_to_inventory", + "ignore", + "non_miner", + }: + reject("discovery_invalid_action", 422) + with self.transaction() as cur: + row, ctx, proof = self.observation(cur, oid, version) + cur.execute( + "SELECT new_value_json FROM audit_logs WHERE entity_id=%s AND action='discovery.reviewed' ORDER BY created_at DESC LIMIT 1", + (oid,), + ) + prior = cur.fetchone() + if prior: + decision = prior["new_value_json"] + if ( + decision["request_version"] == version + and decision["action"] == action + ): + return decision["response"] + reject("discovery_already_reviewed") + if self.observation_version(row) != version: + reject("discovery_version_conflict") + if row["discovery_status"] not in {"new", "known"}: + reject("discovery_not_reviewable") + matched = self.match( + cur, + ctx, + { + "ip_address": str(row["ip_address"]), + "mac_address": row["mac_address"], + "miner_sn": row["miner_sn"], + }, + ) + if self.fingerprint(matched) != proof["miner_fingerprint"]: + reject("discovery_inventory_changed") + mid = matched["id"] if matched else None + state = {"ignore": "ignored", "non_miner": "non_miner"}.get( + action, "known" if mid else "added" + ) + if action == "add_to_inventory" and not mid: + if not row["mac_address"]: + reject("discovery_enrichment_required") + cur.execute( + """INSERT INTO miners (site_id,agent_zone_id,current_mac,current_ip,hostname,miner_sn,model,miner_type,status,notes) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,'discovered','Added by audited observation approval') RETURNING id""", + ( + ctx["site_id"], + ctx["agent_zone_id"], + row["mac_address"], + row["ip_address"], + row["hostname"], + row["miner_sn"], + row["model"], + row["model"], + ), + ) + mid = cur.fetchone()["id"] + cur.execute( + "UPDATE minerops_agent_discoveries SET discovery_status=%s,matched_miner_id=%s,updated_at=clock_timestamp() WHERE id=%s RETURNING *", + (state, mid, oid), + ) + updated = cur.fetchone() + response = { + "ok": True, + "action": action, + "discovery_id": oid, + "miner_id": str(mid) if mid else None, + "version": self.observation_version(updated), + "message": "Discovery decision saved.", + } + self.audit( + cur, + "discovery.reviewed", + oid, + { + "agent_id": ctx["agent_id"], + "agent_zone_id": str(ctx["agent_zone_id"]), + "action": action, + "request_version": version, + "response": response, + }, + user["id"], + self.present(row), + ) + return response + + def token(self, ctx, row): + return jwt.encode( + { + "purpose": "minerops.discovery.miner", + "agent_id": ctx["agent_id"], + "authority": ctx["authority"], + "miner_id": str(row["id"]), + "fingerprint": self.fingerprint(row), + "exp": datetime.now(timezone.utc) + TTL, + }, + settings.jwt_secret_key, + algorithm="HS256", + ) + + def known(self, agent_id): + with self.transaction() as cur: + ctx = self.context(cur, agent_id) + cur.execute( + """SELECT m.*,r.name AS room_name FROM miners m LEFT JOIN rooms r + ON r.id=m.room_id AND r.site_id=m.site_id WHERE m.agent_zone_id=%s + AND m.current_ip IS NOT NULL ORDER BY m.id FOR SHARE OF m""", + (ctx["agent_zone_id"],), + ) + rows = cur.fetchall() + result = [] + for r in rows: + self.in_network(ctx, str(r["current_ip"])) + result.append( + { + "miner_id": str(r["id"]), + "miner_version": self.token(ctx, r), + "ip_address": str(r["current_ip"]), + "mac_address": r["current_mac"], + "room_name": r["room_name"] or ctx["site_name"], + "model": r["model"], + "status": r["status"], + } + ) + return {"ok": True, "count": len(result), "miners": result} + + def bound_miner(self, cur, ctx, item): + object_body(item) + try: + proof = jwt.decode( + item.get("miner_version", ""), + settings.jwt_secret_key, + algorithms=["HS256"], + options={"require": ["exp"]}, + ) + except jwt.PyJWTError: + reject("discovery_miner_version_required") + if ( + proof.get("purpose") != "minerops.discovery.miner" + or proof.get("agent_id") != ctx["agent_id"] + or proof.get("authority") != ctx["authority"] + or proof.get("miner_id") != str(item.get("miner_id")) + ): + reject("discovery_stale_authority") + row = self.match(cur, ctx, item) + if not row or proof.get("fingerprint") != self.fingerprint(row): + reject("discovery_inventory_changed") + return row + + def queue(self, body, user, kind="enrich_match_discovery"): + object_body(body) + with self.transaction() as cur: + if kind == "enrich_match_discovery": + row, ctx, proof = self.observation( + cur, body.get("discovery_id"), body.get("version") + ) + if ( + self.observation_version(row) != body.get("version") + or row["discovery_status"] != "new" + ): + reject("discovery_not_enrichable") + item = { + "ip_address": str(row["ip_address"]), + "mac_address": row["mac_address"], + "miner_sn": row["miner_sn"], + } + matched = self.match(cur, ctx, item) + if self.fingerprint(matched) != proof["miner_fingerprint"]: + reject("discovery_inventory_changed") + payload = { + "discovery_id": row["id"], + "version": body["version"], + "ip_address": str(row["ip_address"]), + } + entity = row["id"] + else: + if not isinstance(body.get("agent_id"), str): + reject("discovery_agent_id_required", 422) + ctx = self.context(cur, body["agent_id"]) + payload = {} + entity = None + if kind != "enrich_match_discovery": + try: + payload["request_id"] = str( + UUID(str(body.get("request_id") or uuid4())) + ) + except (ValueError, TypeError): + reject("discovery_request_id_invalid", 422) + payload.update( + { + "schema_version": 1, + "agent_zone_id": str(ctx["agent_zone_id"]), + "authority": ctx["authority"], + } + ) + # Queue requests are idempotent for the same principal/type/observation version. + qid = str( + uuid5( + NAMESPACE_URL, + "minerops-discovery-command:" + + kind + + ":" + + digest(payload) + + ":" + + str(user["id"]), + ) + ) + cur.execute( + """INSERT INTO minerops_agent_command_queue (id,agent_id,command_type,command_status, + requested_by_user_id,requested_by_username,request_payload) VALUES (%s,%s,%s,'pending',%s,%s,%s) + ON CONFLICT(id) DO NOTHING RETURNING id""", + ( + qid, + ctx["agent_id"], + kind, + str(user["id"]), + user["username"], + Jsonb(payload), + ), + ) + if cur.fetchone(): + self.audit( + cur, + "discovery.command_queued", + entity, + { + "command_id": qid, + "agent_id": ctx["agent_id"], + "type": kind, + "payload": payload, + }, + user["id"], + ) + return { + "ok": True, + "command_id": qid, + "message": "Scoped read command queued.", + } + + def validate_command(self, cur, agent_id, command): + ctx = self.context(cur, agent_id) + p = command["request_payload"] + if ( + not isinstance(p, dict) + or p.get("schema_version") != 1 + or p.get("agent_zone_id") != str(ctx["agent_zone_id"]) + or p.get("authority") != ctx["authority"] + ): + reject("discovery_command_stale_authority") + if command["command_type"] == "enrich_match_discovery": + row, _, proof = self.observation( + cur, p.get("discovery_id"), p.get("version") + ) + if ( + row["agent_id"] != agent_id + or self.observation_version(row) != p["version"] + or row["discovery_status"] != "new" + or str(row["ip_address"]) != p.get("ip_address") + ): + reject("discovery_command_stale_observation") + if ( + self.fingerprint( + self.match( + cur, + ctx, + { + "ip_address": p["ip_address"], + "mac_address": row["mac_address"], + "miner_sn": row["miner_sn"], + }, + ) + ) + != proof["miner_fingerprint"] + ): + reject("discovery_inventory_changed") + return ctx + + def target(self, agent_id, body): + self.identity(body, agent_id) + try: + UUID(str(body.get("command_id"))) + except (ValueError, TypeError): + reject("discovery_command_id_required", 422) + with self.transaction() as cur: + cur.execute( + "SELECT * FROM minerops_agent_command_queue WHERE id=%s AND agent_id=%s FOR UPDATE", + (body.get("command_id"), agent_id), + ) + command = cur.fetchone() + if ( + not command + or command["command_status"] != "claimed" + or command["command_type"] != "enrich_match_discovery" + ): + reject("discovery_enrichment_command_required") + self.validate_command(cur, agent_id, command) + p = command["request_payload"] + if p["discovery_id"] != body.get("discovery_id") or p[ + "version" + ] != body.get("version"): + reject("discovery_command_observation_conflict") + return {"ok": True, "ip_address": p["ip_address"]} + + def enrich(self, agent_id, body): + self.identity(body, agent_id) + with self.transaction() as cur: + ctx = self.context(cur, agent_id) + if body.get("discovery_id"): + try: + UUID(str(body.get("command_id"))) + except (ValueError, TypeError): + reject("discovery_command_id_required", 422) + row, _, proof = self.observation( + cur, body["discovery_id"], body.get("version") + ) + if ( + row["agent_id"] != agent_id + or row["discovery_status"] != "new" + or body.get("version") != self.observation_version(row) + ): + reject("discovery_not_enrichable") + cur.execute( + "SELECT * FROM minerops_agent_command_queue WHERE id=%s AND agent_id=%s FOR UPDATE", + (body.get("command_id"), agent_id), + ) + command = cur.fetchone() + if ( + not command + or command["command_status"] != "claimed" + or command["command_type"] != "enrich_match_discovery" + ): + reject("discovery_enrichment_command_required") + self.validate_command(cur, agent_id, command) + if command["request_payload"]["discovery_id"] != row["id"]: + reject("discovery_command_observation_conflict") + if address(body.get("ip_address")) != str(row["ip_address"]): + reject("discovery_identity_conflict") + item = { + k: body.get(k) + for k in ( + "ip_address", + "mac_address", + "model", + "hostname", + "miner_sn", + ) + } + item["parent_observation_id"] = row["id"] + item["command_id"] = command["id"] + child = self.observed(cur, ctx, item, parent=row["id"]) + return { + "ok": True, + "matched": bool(child["matched_miner_id"]), + "discovery": self.present(child), + "message": "Enrichment appended; review the new observation.", + } + row = self.bound_miner(cur, ctx, body) + changes = { + k: text(body.get(k)) + for k in ( + "hostname", + "model", + "miner_type", + "api_version", + "firmware_version", + "platform", + ) + } + changes = {k: v for k, v in changes.items() if v is not None} + if changes: + sql = ", ".join(k + "=%s" for k in changes) + cur.execute( + "UPDATE miners SET " + + sql + + ",updated_at=clock_timestamp() WHERE id=%s AND agent_zone_id=%s RETURNING *", + (*changes.values(), row["id"], ctx["agent_zone_id"]), + ) + updated = cur.fetchone() + else: + updated = row + self.audit( + cur, + "discovery.miner_enriched", + row["id"], + { + "agent_id": agent_id, + "changes": changes, + "raw_version": body.get("raw_version"), + "raw_miner_info": body.get("raw_miner_info"), + }, + old={"fingerprint": self.fingerprint(row)}, + ) + return { + "ok": True, + "matched": True, + "miner": { + k: ( + str(v) + if isinstance( + v, + ( + UUID, + datetime, + ipaddress.IPv4Address, + ipaddress.IPv6Address, + ), + ) + else v + ) + for k, v in updated.items() + }, + "message": "Enrichment saved without changing ownership or history.", + } + + def snapshots(self, agent_id, body, kind="miner"): + self.identity(body, agent_id) + items = ( + body.get("miners") + if kind == "miner" + else body.get("pool_snapshots", body.get("snapshots")) + ) + if not isinstance(items, list) or len(items) > MAX_BATCH: + reject("discovery_invalid_batch", 422) + with self.transaction() as cur: + ctx = self.context(cur, agent_id) + for item in items: + row = self.bound_miner(cur, ctx, item) + if kind == "miner": + if type(item.get("online")) is not bool: + reject("discovery_invalid_online", 422) + status = "online" if item["online"] else "offline" + cur.execute( + "UPDATE miners SET status=%s,last_seen_at=CASE WHEN %s THEN clock_timestamp() ELSE last_seen_at END,updated_at=clock_timestamp() WHERE id=%s AND agent_zone_id=%s", + (status, item["online"], row["id"], ctx["agent_zone_id"]), + ) + summary = object_body( + object_body(item.get("responses") or {}).get("summary") or {} + ) + raw = summary.get("json") + if raw is None: + try: + raw = json.loads(summary.get("raw") or "{}") + except (ValueError, TypeError): + reject("discovery_invalid_summary", 422) + status_snapshot(cur, row, raw) + else: + pools = item.get("pools", []) + if not isinstance(pools, list) or len(pools) > 32: + reject("discovery_invalid_pools", 422) + for p in pools: + object_body(p) + pool_snapshot(cur, row, p, agent_id) + self.audit( + cur, + "discovery." + kind + "_snapshot", + row["id"], + {"agent_id": agent_id, "payload": item}, + ) + if kind == "miner": + cur.execute( + """INSERT INTO minerops_agent_snapshots(agent_id,agent_zone_id,site_name,miner_count,payload) + VALUES (%s,%s,%s,%s,%s) RETURNING id,received_at""", + ( + agent_id, + ctx["agent_zone_id"], + ctx["site_name"], + len(items), + Jsonb(body), + ), + ) + receipt = cur.fetchone() + else: + receipt = {} + return { + "ok": True, + "applied": len(items), + "inserted": len(items), + "matched": len(items), + "miner_count": len(items), + "snapshot_id": receipt.get("id"), + "message": "Scoped telemetry appended.", + } + + def api3(self, agent_id, body): + self.identity(body, agent_id) + items = body.get("items") + if not isinstance(items, list) or len(items) > MAX_BATCH: + reject("discovery_invalid_batch", 422) + with self.transaction() as cur: + ctx = self.context(cur, agent_id) + counts = dict( + processed=len(items), + matched=len(items), + api3_ok=0, + api3_failed=0, + miner_updates=0, + status_rows=0, + pool_rows=0, + hashboard_rows=0, + ) + for item in items: + object_body(item) + device = object_body(item.get("device_info") or {}) + message = object_body(device.get("msg") or {}) + network = object_body(message.get("network") or {}) + if network.get("ip") and address(network["ip"]) != address( + item.get("ip") + ): + reject("discovery_identity_conflict") + if network.get("mac") and mac(network["mac"]) != mac(item.get("mac")): + reject("discovery_identity_conflict") + row = self.bound_miner(cur, ctx, item) + if type(item.get("api3_supported")) is not bool: + reject("discovery_invalid_api3_supported", 422) + miner = object_body(message.get("miner") or {}) + system = object_body(message.get("system") or {}) + power = object_body(message.get("power") or {}) + reported_serial = text(miner.get("miner-sn")) + if ( + reported_serial + and row["miner_sn"] + and reported_serial != row["miner_sn"] + ): + reject("discovery_identity_conflict") + metadata = { + "model": text(miner.get("type")), + "miner_type": text(miner.get("type")), + "api_version": text(system.get("api")), + "api3_api_version": text(system.get("api")), + "api3_api_switch": text(system.get("apiswitch")), + "firmware_version": text(system.get("fwversion")), + "platform": text(system.get("platform")), + } + metadata = {k: v for k, v in metadata.items() if v is not None} + if metadata: + assignments = [ + k + + ( + "=COALESCE(NULLIF(" + k + ",''),%s)" + if k in {"model", "miner_type"} + else "=%s" + ) + for k in metadata + ] + cur.execute( + "UPDATE miners SET " + + ",".join(assignments) + + " WHERE id=%s AND agent_zone_id=%s", + (*metadata.values(), row["id"], ctx["agent_zone_id"]), + ) + counts["api3_ok" if item["api3_supported"] else "api3_failed"] += 1 + counts["miner_updates"] += 1 + cur.execute( + "UPDATE miners SET api3_supported=%s,api3_last_seen_at=clock_timestamp(),api3_last_error=%s,updated_at=clock_timestamp() WHERE id=%s AND agent_zone_id=%s", + ( + item["api3_supported"], + text(item.get("api3_error")), + row["id"], + ctx["agent_zone_id"], + ), + ) + cur.execute( + """INSERT INTO miner_api3_enrichment_snapshots + (miner_id,agent_id,site_name,ip_address,mac_address,api_supported,api_error,device_info,miner_setting,miner_status,fan_setting,custom_data,api_version,api_switch,model,miner_sn,power_sn,psu_model) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", + ( + row["id"], + agent_id, + ctx["site_name"], + row["current_ip"], + row["current_mac"], + item["api3_supported"], + text(item.get("api3_error")), + *[ + Jsonb(item.get(k)) + for k in ( + "device_info", + "miner_setting", + "miner_status", + "fan_setting", + "custom_data", + ) + ], + metadata.get("api_version"), + metadata.get("api3_api_switch"), + metadata.get("model"), + reported_serial, + text(power.get("sn")), + text(power.get("model")), + ), + ) + status = object_body(item.get("miner_status") or {}) + status_msg = object_body(status.get("msg") or {}) + summary = object_body(status_msg.get("summary") or {}) + if item["api3_supported"] and summary: + status_snapshot(cur, row, status, api3=True) + counts["status_rows"] += 1 + pools = status_msg.get("pools") or [] + boards = status_msg.get("edevs") or [] + if ( + not isinstance(pools, list) + or len(pools) > 32 + or not isinstance(boards, list) + or len(boards) > 32 + ): + reject("discovery_invalid_telemetry_batch", 422) + for pool in pools: + pool_snapshot(cur, row, pool, agent_id) + counts["pool_rows"] += 1 + for board in boards: + object_body(board) + values = { + "miner_id": row["id"], + "ip_address": row["current_ip"], + "mac_address": row["current_mac"], + "raw_hashboard_json": Jsonb(board), + } + for key, column in { + "slot": "slot", + "hash-average": "hash_average_ths", + "factory-hash": "factory_hash_ths", + "freq": "freq", + "effective-chips": "effective_chips", + "chip-temp-min": "chip_temp_min_c", + "chip-temp-avg": "chip_temp_avg_c", + "chip-temp-max": "chip_temp_max_c", + }.items(): + if key in board: + values[column] = number( + board[key], column in {"slot", "effective_chips"} + ) + insert(cur, "miner_api3_hashboard_snapshots", values) + counts["hashboard_rows"] += 1 + self.audit( + cur, + "discovery.api3_snapshot", + row["id"], + {"agent_id": agent_id, "payload": item}, + ) + return { + "data": {"ok": True, **counts}, + "meta": {}, + "error": None, + } diff --git a/backend/app/identity.py b/backend/app/identity.py index e888ad4..b9ea845 100644 --- a/backend/app/identity.py +++ b/backend/app/identity.py @@ -711,6 +711,7 @@ def machine_agent_request(path: str) -> bool: "/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", @@ -826,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" diff --git a/backend/app/main.py b/backend/app/main.py index a0d067e..17fff38 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,3 +1,4 @@ +from app.discovery.http import dispatch as discovery_dispatch from datetime import datetime, timedelta, timezone from typing import Any from uuid import UUID @@ -2137,799 +2138,27 @@ def find_scan_range_context(cur, ip_address: str, scan_range_id: str | None) -> def find_existing_miner(cur, miner_sn: str | None, mac_address: str | None, ip_address: str) -> dict | None: - if miner_sn: - cur.execute( - """ - SELECT id::text - FROM miners - WHERE miner_sn = %s - LIMIT 1; - """, - (miner_sn,), - ) - row = cur.fetchone() - if row: - return dict(row) - - if mac_address: - cur.execute( - """ - SELECT id::text - FROM miners - WHERE lower(current_mac) = lower(%s) - LIMIT 1; - """, - (mac_address,), - ) - row = cur.fetchone() - if row: - return dict(row) - - cur.execute( - """ - SELECT id::text - FROM miners - WHERE current_ip = %s - LIMIT 1; - """, - (ip_address,), + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - row = cur.fetchone() - - if row: - return dict(row) - - return None @app.post("/api/miners/discover-one") def discover_one_miner(request: DiscoverOneMinerRequest, current_user: dict = Depends(get_current_user)): - try: - ip_address = str(ipaddress.ip_address(request.ip_address)) - except ValueError as exc: - raise HTTPException(status_code=422, detail=f"Invalid IP address: {exc}") - - miner_info_result = send_whatsminer_read_command( - ip_address=ip_address, - port=request.port, - command="get_miner_info", - timeout_ms=request.timeout_ms, - params={ - "info": "ip,proto,netmask,gateway,dns,hostname,mac,ledstat,minersn,powersn" - }, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - version_result = send_whatsminer_read_command( - ip_address=ip_address, - port=request.port, - command="get_version", - timeout_ms=request.timeout_ms, - ) - - summary_result = send_whatsminer_read_command( - ip_address=ip_address, - port=request.port, - command="summary", - timeout_ms=request.timeout_ms, - ) - - edevs_result = send_whatsminer_read_command( - ip_address=ip_address, - port=request.port, - command="edevs", - timeout_ms=request.timeout_ms, - ) - - pools_result = send_whatsminer_read_command( - ip_address=ip_address, - port=request.port, - command="pools", - timeout_ms=request.timeout_ms, - ) - - miner_info_json = miner_info_result.get("parsed_json") or {} - version_json = version_result.get("parsed_json") or {} - summary_json = summary_result.get("parsed_json") or {} - edevs_json = edevs_result.get("parsed_json") or {} - pools_json = pools_result.get("parsed_json") or {} - - miner_info_msg = get_msg_dict(miner_info_result, "get_miner_info") - version_msg = get_msg_dict(version_result, "get_version") - try: - summary_msg = get_msg_dict(summary_result, "summary") - summary_warning = None - except HTTPException as exc: - summary_msg = {} - summary_warning = exc.detail - - mac_address = miner_info_msg.get("mac") - if mac_address: - mac_address = str(mac_address).upper() - - hostname = miner_info_msg.get("hostname") - miner_sn = miner_info_msg.get("minersn") - power_sn = miner_info_msg.get("powersn") - - miner_type = version_msg.get("miner_type") - model = miner_type.split("_")[0] if miner_type else None - - conn = None - - try: - conn = get_db_connection() - conn.autocommit = False - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - context = find_scan_range_context(cur, ip_address, request.scan_range_id) - - existing_miner = find_existing_miner( - cur=cur, - miner_sn=miner_sn, - mac_address=mac_address, - ip_address=ip_address, - ) - - if existing_miner: - miner_id = existing_miner["id"] - - cur.execute( - """ - UPDATE miners - SET site_id = COALESCE(%s, site_id), - room_id = COALESCE(%s, room_id), - current_mac = %s, - current_ip = %s, - hostname = %s, - miner_sn = %s, - power_sn = %s, - model = %s, - miner_type = %s, - api_version = %s, - firmware_version = %s, - platform = %s, - chip = %s, - status = 'online', - last_seen_at = now(), - updated_at = now() - WHERE id = %s - RETURNING id::text; - """, - ( - context.get("site_id"), - context.get("room_id"), - mac_address, - ip_address, - hostname, - miner_sn, - power_sn, - model, - miner_type, - version_msg.get("api_ver"), - version_msg.get("fw_ver"), - version_msg.get("platform"), - version_msg.get("chip"), - miner_id, - ), - ) - miner_id = cur.fetchone()["id"] - action = "miner.updated" - - else: - cur.execute( - """ - INSERT INTO miners ( - site_id, - room_id, - current_mac, - current_ip, - hostname, - miner_sn, - power_sn, - model, - miner_type, - api_version, - firmware_version, - platform, - chip, - status - ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'online') - RETURNING id::text; - """, - ( - context.get("site_id"), - context.get("room_id"), - mac_address, - ip_address, - hostname, - miner_sn, - power_sn, - model, - miner_type, - version_msg.get("api_ver"), - version_msg.get("fw_ver"), - version_msg.get("platform"), - version_msg.get("chip"), - ), - ) - miner_id = cur.fetchone()["id"] - action = "miner.created" - - if mac_address: - cur.execute( - """ - UPDATE miner_network_identities - SET is_current = false, - updated_at = now() - WHERE miner_id = %s; - """, - (miner_id,), - ) - - cur.execute( - """ - SELECT id::text - FROM miner_network_identities - WHERE miner_id = %s - AND lower(mac_address) = lower(%s) - LIMIT 1; - """, - (miner_id, mac_address), - ) - identity = cur.fetchone() - - if identity: - cur.execute( - """ - UPDATE miner_network_identities - SET ip_address = %s, - hostname = %s, - source = 'scanner', - is_current = true, - last_seen_at = now(), - updated_at = now() - WHERE id = %s; - """, - (ip_address, hostname, identity["id"]), - ) - else: - cur.execute( - """ - INSERT INTO miner_network_identities ( - miner_id, - mac_address, - ip_address, - hostname, - source, - is_current - ) - VALUES (%s, %s, %s, %s, 'scanner', true); - """, - (miner_id, mac_address, ip_address, hostname), - ) - - for board in edevs_json.get("DEVS", []): - slot = board.get("Slot") - - if slot is None: - continue - - cur.execute( - """ - INSERT INTO miner_hashboards ( - miner_id, - slot, - pcb_sn, - chip_data, - temperature_c, - chip_frequency, - mhs_av, - mhs_1m, - mhs_15m, - hs_rt, - factory_ghs, - effective_chips, - upfreq_complete, - raw_json - ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) - ON CONFLICT ON CONSTRAINT uq_miner_hashboards_miner_slot - DO UPDATE SET - pcb_sn = EXCLUDED.pcb_sn, - chip_data = EXCLUDED.chip_data, - temperature_c = EXCLUDED.temperature_c, - chip_frequency = EXCLUDED.chip_frequency, - mhs_av = EXCLUDED.mhs_av, - mhs_1m = EXCLUDED.mhs_1m, - mhs_15m = EXCLUDED.mhs_15m, - hs_rt = EXCLUDED.hs_rt, - factory_ghs = EXCLUDED.factory_ghs, - effective_chips = EXCLUDED.effective_chips, - upfreq_complete = EXCLUDED.upfreq_complete, - raw_json = EXCLUDED.raw_json, - last_seen_at = now(), - updated_at = now(); - """, - ( - miner_id, - int(slot), - board.get("PCB SN"), - board.get("Chip Data"), - board.get("Temperature"), - board.get("Chip Frequency"), - board.get("MHS av"), - board.get("MHS 1m"), - board.get("MHS 15m"), - board.get("HS RT"), - board.get("Factory GHS"), - board.get("Effective Chips"), - board.get("Upfreq Complete"), - psycopg.types.json.Jsonb(json_safe(board)), - ), - ) - - cur.execute( - """ - INSERT INTO miner_status_snapshots ( - miner_id, - observed_ip, - mhs_av, - mhs_1m, - mhs_15m, - hs_rt, - power_w, - power_rate_w_per_th, - env_temp_c, - chip_temp_min_c, - chip_temp_max_c, - chip_temp_avg_c, - fan_speed_in_rpm, - fan_speed_out_rpm, - uptime_seconds, - hash_stable, - pool_rejected_percent, - raw_summary_json, - raw_version_json, - raw_miner_info_json - ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s); - """, - ( - miner_id, - ip_address, - summary_msg.get("MHS av"), - summary_msg.get("MHS 1m"), - summary_msg.get("MHS 15m"), - summary_msg.get("HS RT"), - summary_msg.get("Power"), - summary_msg.get("Power Rate"), - summary_msg.get("Env Temp"), - summary_msg.get("Chip Temp Min"), - summary_msg.get("Chip Temp Max"), - summary_msg.get("Chip Temp Avg"), - summary_msg.get("Fan Speed In"), - summary_msg.get("Fan Speed Out"), - summary_msg.get("Uptime"), - to_bool_or_none(summary_msg.get("Hash Stable")), - summary_msg.get("Pool Rejected%"), - psycopg.types.json.Jsonb(json_safe(summary_json)), - psycopg.types.json.Jsonb(json_safe(version_json)), - psycopg.types.json.Jsonb(json_safe(miner_info_json)), - ), - ) - - cur.execute( - """ - INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json) - VALUES (%s, %s, 'miner', %s, %s); - """, - ( - current_user["id"], - action, - miner_id, - psycopg.types.json.Jsonb( - { - "ip_address": ip_address, - "mac_address": mac_address, - "miner_sn": miner_sn, - "power_sn": power_sn, - "miner_type": miner_type, - "firmware_version": version_msg.get("fw_ver"), - "room_id": context.get("room_id"), - "scan_range_id": context.get("scan_range_id"), - } - ), - ), - ) - - cur.execute( - """ - SELECT - miners.id::text, - miners.site_id::text, - sites.name AS site_name, - miners.room_id::text, - rooms.name AS room_name, - miners.current_mac, - host(miners.current_ip) AS current_ip, - miners.hostname, - miners.miner_sn, - miners.power_sn, - miners.model, - miners.miner_type, - miners.api_version, - miners.firmware_version, - miners.platform, - miners.chip, - miners.status, - miners.first_seen_at, - miners.last_seen_at, - miners.updated_at - FROM miners - LEFT JOIN sites ON sites.id = miners.site_id - LEFT JOIN rooms ON rooms.id = miners.room_id - WHERE miners.id = %s; - """, - (miner_id,), - ) - miner = dict(cur.fetchone()) - - conn.commit() - - return api_response( - data={ - "action": action, - "miner": miner, - "summary": { - "ip_address": ip_address, - "mac_address": mac_address, - "miner_sn": miner_sn, - "power_sn": power_sn, - "miner_type": miner_type, - "firmware_version": version_msg.get("fw_ver"), - "hashboards_seen": len(edevs_json.get("DEVS", [])), - "pools_seen": len(pools_json.get("POOLS", [])), - "hashrate_mhs_av": summary_msg.get("MHS av"), - "power_w": summary_msg.get("Power"), - "chip_temp_max_c": summary_msg.get("Chip Temp Max"), - "summary_warning": summary_warning, - }, - } - ) - - except HTTPException: - if conn is not None: - conn.rollback() - raise - except Exception as exc: - if conn is not None: - conn.rollback() - raise HTTPException(status_code=500, detail=str(exc)) - finally: - if conn is not None: - conn.close() - @app.post("/api/scan-ranges/{scan_range_id}/discover-now") def discover_scan_range_now(scan_range_id: str, request: ScanNowRequest, current_user: dict = Depends(get_current_user)): - conn = None - - try: - conn = get_db_connection() - conn.autocommit = False - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - cur.execute( - """ - SELECT - id::text, - room_id::text, - name, - range_type, - cidr, - host(start_ip) AS start_ip, - host(end_ip) AS end_ip, - enabled, - timeout_ms, - retry_limit, - parallelism_limit, - stagger_ms, - api_port, - scan_interval_seconds - FROM scan_ranges - WHERE id = %s; - """, - (scan_range_id,), - ) - scan_range = cur.fetchone() - - if not scan_range: - raise HTTPException(status_code=404, detail="Scan range not found") - - if not scan_range["enabled"]: - raise HTTPException(status_code=409, detail="Scan range is disabled") - - ip_list = expand_scan_range_for_scan(dict(scan_range), request.max_ips, request.specific_ips) - - cur.execute( - """ - INSERT INTO scan_runs ( - scan_range_id, - triggered_by, - triggered_by_user_id, - status, - total_ips, - scanned_ips, - miners_found, - errors_count, - notes - ) - VALUES (%s, 'manual', %s, 'running', %s, 0, 0, 0, %s) - RETURNING id::text; - """, - ( - scan_range_id, - current_user["id"], - len(ip_list), - "TCP port 4028 scan with read-only miner discovery", - ), - ) - scan_run_id = cur.fetchone()["id"] - - cur.execute( - """ - INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json) - VALUES (%s, 'scan.discovery.started', 'scan_run', %s, %s); - """, - ( - current_user["id"], - scan_run_id, - psycopg.types.json.Jsonb( - { - "scan_range_id": scan_range_id, - "scan_range_name": scan_range["name"], - "total_ips": len(ip_list), - "mode": "tcp_4028_then_whatsminer_read_only_discovery", - } - ), - ), - ) - - conn.commit() - - except HTTPException: - if conn is not None: - conn.rollback() - raise - except Exception as exc: - if conn is not None: - conn.rollback() - raise HTTPException(status_code=500, detail=str(exc)) - finally: - if conn is not None: - conn.close() - - results = [] - detected_count = 0 - - max_workers = min(scan_range["parallelism_limit"], len(ip_list)) - if max_workers < 1: - max_workers = 1 - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - future_map = { - executor.submit( - tcp_check_ip, - ip, - scan_range["api_port"], - scan_range["timeout_ms"], - ): ip - for ip in ip_list - } - - for future in as_completed(future_map): - result = future.result() - results.append(result) - - if result["detected"]: - detected_count += 1 - - results = sorted(results, key=lambda item: ipaddress.ip_address(item["ip_address"])) - - conn = None - - try: - conn = get_db_connection() - conn.autocommit = False - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - for result in results: - cur.execute( - """ - INSERT INTO scan_results ( - scan_run_id, - ip_address, - port, - detected, - response_ms, - status, - error_message - ) - VALUES (%s, %s, %s, %s, %s, %s, %s); - """, - ( - scan_run_id, - result["ip_address"], - result["port"], - result["detected"], - result["response_ms"], - result["status"], - result["error_message"], - ), - ) - - cur.execute( - """ - UPDATE scan_ranges - SET last_scan_at = now(), - next_scan_at = now() + make_interval(secs => scan_interval_seconds), - updated_at = now() - WHERE id = %s; - """, - (scan_range_id,), - ) - - cur.execute( - """ - UPDATE scan_runs - SET scanned_ips = %s, - miners_found = %s, - errors_count = 0 - WHERE id = %s; - """, - ( - len(results), - detected_count, - scan_run_id, - ), - ) - - conn.commit() - - except Exception as exc: - if conn is not None: - conn.rollback() - raise HTTPException(status_code=500, detail=str(exc)) - finally: - if conn is not None: - conn.close() - - discoveries = [] - discovery_errors = [] - - for result in results: - if not result["detected"]: - continue - - try: - discovery_response = discover_one_miner( - DiscoverOneMinerRequest( - ip_address=result["ip_address"], - port=result["port"], - scan_range_id=scan_range_id, - timeout_ms=scan_range["timeout_ms"], - ), - current_user=current_user, - ) - - discoveries.append(discovery_response["data"]) - - except HTTPException as exc: - discovery_errors.append( - { - "ip_address": result["ip_address"], - "status_code": exc.status_code, - "detail": exc.detail, - } - ) - - except Exception as exc: - discovery_errors.append( - { - "ip_address": result["ip_address"], - "status_code": 500, - "detail": str(exc), - } - ) - - final_status = "completed" - if discovery_errors: - final_status = "completed_with_errors" - - conn = None - - try: - conn = get_db_connection() - conn.autocommit = False - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - cur.execute( - """ - UPDATE scan_runs - SET finished_at = now(), - status = %s, - miners_found = %s, - errors_count = %s, - notes = %s - WHERE id = %s - RETURNING - id::text, - scan_range_id::text, - triggered_by, - started_at, - finished_at, - status, - total_ips, - scanned_ips, - miners_found, - errors_count, - notes; - """, - ( - final_status, - len(discoveries), - len(discovery_errors), - "TCP port 4028 scan with read-only miner discovery", - scan_run_id, - ), - ) - scan_run = cur.fetchone() - - cur.execute( - """ - INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json) - VALUES (%s, 'scan.discovery.completed', 'scan_run', %s, %s); - """, - ( - current_user["id"], - scan_run_id, - psycopg.types.json.Jsonb( - { - "scan_range_id": scan_range_id, - "total_ips": len(ip_list), - "scanned_ips": len(results), - "port_4028_open": detected_count, - "miners_discovered": len(discoveries), - "discovery_errors": len(discovery_errors), - } - ), - ), - ) - - conn.commit() - - except Exception as exc: - if conn is not None: - conn.rollback() - raise HTTPException(status_code=500, detail=str(exc)) - finally: - if conn is not None: - conn.close() - - return api_response( - data={ - "scan_run": dict(scan_run), - "summary": { - "scan_range_id": scan_range_id, - "scan_range_name": scan_range["name"], - "total_ips": len(ip_list), - "scanned_ips": len(results), - "port_4028_open": detected_count, - "miners_discovered": len(discoveries), - "discovery_errors": len(discovery_errors), - }, - "discoveries": discoveries, - "discovery_errors": discovery_errors, - } + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) @app.get("/api/dashboard/summary") @@ -3101,6 +2330,7 @@ def model_from_miner_type(miner_type: str | None) -> str | None: async def import_whatsminer_tool_scan( dry_run: bool = Query(default=True), update_existing: bool = Query(default=False), + agent_zone_id: str | None = Query(default=None), file: UploadFile = File(...), current_user: dict = Depends(get_current_user), ): @@ -3112,6 +2342,8 @@ async def import_whatsminer_tool_scan( csv_text = file_bytes.decode("latin-1") reader = csv.DictReader(io.StringIO(csv_text)) + if reader.fieldnames and len(set(reader.fieldnames)) != len(reader.fieldnames): + raise HTTPException(status_code=422, detail="CSV contains duplicate column names") rows = list(reader) if not rows: @@ -3124,6 +2356,7 @@ async def import_whatsminer_tool_scan( skipped_no_useful_data = 0 update_samples = [] + audited_updates = [] unmatched_samples = [] conn = None @@ -3133,6 +2366,12 @@ async def import_whatsminer_tool_scan( conn.autocommit = False with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: + from app.discovery.csv_import import resolve_inventory_row + from app.discovery.repository import DiscoveryRepository + + # Same lock order as discovery; includes noncooperating inventory writers. + # Selection, metadata changes and audit stay in this one transaction. + DiscoveryRepository.lock(cur) for row_number, row in enumerate(rows, start=2): ip_address = csv_value(row, ["IP", "IP Address", "ip", "ip_address"]) mac_address = csv_value(row, ["MAC Addr", "MAC", "Mac", "mac", "mac_address"]) @@ -3142,50 +2381,8 @@ async def import_whatsminer_tool_scan( if ip_address: ip_address = clean_ip_text(ip_address) - normalized_mac = normalize_mac_for_match(mac_address) model = model_from_miner_type(miner_type) - - if not ip_address and not normalized_mac: - skipped_no_useful_data += 1 - continue - - miner = None - - if normalized_mac: - cur.execute( - """ - SELECT - id::text, - host(current_ip) AS current_ip, - current_mac, - model, - miner_type, - firmware_version - FROM miners - WHERE regexp_replace(lower(current_mac), '[^0-9a-f]', '', 'g') = %s - LIMIT 1; - """, - (normalized_mac,), - ) - miner = cur.fetchone() - - if not miner and ip_address: - cur.execute( - """ - SELECT - id::text, - host(current_ip) AS current_ip, - current_mac, - model, - miner_type, - firmware_version - FROM miners - WHERE host(current_ip) = %s - LIMIT 1; - """, - (ip_address,), - ) - miner = cur.fetchone() + miner = resolve_inventory_row(cur, row, row_number, agent_zone_id) if not miner: unmatched_count += 1 @@ -3263,6 +2460,11 @@ async def import_whatsminer_tool_scan( tuple(params), ) + audited_updates.append({ + "miner_id": miner["id"], "site_id": miner["site_id"], + "agent_zone_id": miner["agent_zone_id"], + "changes": proposed_changes, + }) updates_applied += 1 if not dry_run: @@ -3282,6 +2484,8 @@ async def import_whatsminer_tool_scan( "rows_with_updates": rows_with_updates, "updates_applied": updates_applied, "update_existing": update_existing, + "agent_zone_id": agent_zone_id, + "updates": audited_updates, } ), ), @@ -3475,6 +2679,9 @@ def whatsminer_import_tool_page():

2. Upload CSV

+

Provide a Miner ID, current MAC, or serial number in each row. For IP-only rows, select a Zone below or include an Agent Zone ID column. All supplied identifiers must agree with current inventory.

+ + @@ -3665,7 +2872,9 @@ async function runImport(dryRun) { const formData = new FormData(); formData.append("file", file); - const url = "/api/imports/whatsminer-tool-scan?dry_run=" + dryRun + "&update_existing=" + updateExisting; + const zoneId = document.getElementById("importZoneId").value.trim(); + const url = "/api/imports/whatsminer-tool-scan?dry_run=" + dryRun + "&update_existing=" + updateExisting + + (zoneId ? "&agent_zone_id=" + encodeURIComponent(zoneId) : ""); const response = await fetch(url, { method: "POST", @@ -3984,192 +3193,7 @@ async def agent_pool_snapshots( authenticated_agent_id: str = Depends(require_agent_identity), zone_service: AgentZoneService = Depends(get_agent_zone_service), ): - import json as _json - import uuid as _uuid - - routing = zone_service.routing_context(authenticated_agent_id) - if routing.routing_reason or routing.agent_zone_id is None: - raise HTTPException( - status_code=409, - detail=( - "Agent routing unavailable: " - + (routing.routing_reason or "agent_zone_unassigned") - ), - ) - - body = await request.json() - - agent_id = authenticated_agent_id - snapshots = body.get("pool_snapshots") or body.get("snapshots") or [] - - minerops_ensure_pool_info_table() - - inserted = 0 - matched = 0 - - conn = None - - def fnum(value): - try: - if value is None or value == "": - return None - return float(value) - except Exception: - return None - - def fint(value): - try: - if value is None or value == "": - return None - return int(float(value)) - except Exception: - return None - - try: - conn = get_db_connection() - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - for snap in snapshots: - ip = (snap.get("ip_address") or snap.get("ip") or "").strip() - mac = (snap.get("mac_address") or snap.get("mac") or "").strip() - model = snap.get("model") or "" - pools = snap.get("pools") or [] - - if isinstance(pools, dict): - pools = [pools] - - if not isinstance(pools, list): - pools = [] - - if not ip or not pools: - continue - - miner_id = None - - if mac: - cur.execute( - """ - SELECT id - FROM miners - WHERE agent_zone_id = %s - AND ( - upper(current_mac) = upper(%s) - OR host(current_ip) = %s - ) - LIMIT 1; - """, - (routing.agent_zone_id, mac, ip), - ) - else: - cur.execute( - """ - SELECT id - FROM miners - WHERE agent_zone_id = %s - AND host(current_ip) = %s - LIMIT 1; - """, - (routing.agent_zone_id, ip), - ) - - row = cur.fetchone() - - if not row: - continue - - miner_id = row["id"] - matched += 1 - - for pool in pools: - if not isinstance(pool, dict): - continue - - pool_index = fint(pool.get("POOL") or pool.get("pool") or pool.get("Pool")) - - cur.execute( - """ - INSERT INTO miner_pool_snapshots ( - miner_id, - ip_address, - mac_address, - model, - pool_index, - url, - worker_user, - pool_status, - stratum_active, - accepted, - rejected, - stale, - get_failures, - remote_failures, - pool_rejected_percent, - pool_stale_percent, - last_share_time_epoch, - raw_pool_json, - agent_id - ) - VALUES ( - %s, - %s::inet, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s, - %s::jsonb, - %s - ); - """, - ( - miner_id, - ip, - mac, - model, - pool_index, - pool.get("URL") or pool.get("url"), - pool.get("User") or pool.get("user"), - pool.get("Status") or pool.get("status"), - pool.get("Stratum Active") if "Stratum Active" in pool else pool.get("stratum_active"), - fnum(pool.get("Accepted")), - fnum(pool.get("Rejected")), - fnum(pool.get("Stale")), - fnum(pool.get("Get Failures")), - fnum(pool.get("Remote Failures")), - fnum(pool.get("Pool Rejected%")), - fnum(pool.get("Pool Stale%")), - fnum(pool.get("Last Share Time")), - _json.dumps(pool), - agent_id, - ), - ) - - inserted += 1 - - conn.commit() - - return api_response( - data={ - "inserted": inserted, - "matched": matched, - "received_miners": len(snapshots), - }, - meta={"message": "Pool snapshots saved."}, - ) - - finally: - if conn is not None: - conn.close() + return await discovery_dispatch('snapshots', request, agent_id=authenticated_agent_id, kind='pool') @app.get("/api/dashboard/pool-info-latest") @@ -13616,360 +12640,86 @@ def get_known_miners_for_scheduled_scan(): def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - signature = inspect.signature(func) - params = signature.parameters - - candidate_kwargs = { - "ip_address": ip_address, - "ip": ip_address, - "host": ip_address, - "room_id": room_id, - "source": "scheduled_known_miner_scan", - } - - kwargs = {} - - for name, param in params.items(): - if name in candidate_kwargs: - kwargs[name] = candidate_kwargs[name] - - missing_required = [] - - for name, param in params.items(): - if param.default is inspect._empty and name not in kwargs: - if param.kind in ( - inspect.Parameter.POSITIONAL_OR_KEYWORD, - inspect.Parameter.KEYWORD_ONLY, - ): - missing_required.append(name) - - if missing_required: - raise RuntimeError( - "discover_one_miner requires unsupported argument(s): " - + ", ".join(missing_required) - ) - - result = func(**kwargs) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) def refresh_one_known_miner_for_schedule(miner_row: dict): - ip_address = miner_row.get("ip_address") - room_id = miner_row.get("room_id") - - if not ip_address: - raise RuntimeError("Miner has no IP address") - - call_discover_one_miner_flexible(ip_address=ip_address, room_id=room_id) - - return { - "miner_id": miner_row.get("miner_id"), - "ip_address": ip_address, - "mac": miner_row.get("current_mac"), - "room_name": miner_row.get("room_name"), - } + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) def run_scheduled_known_miner_scan(reason: str = "scheduled"): - if not SCHEDULED_SCANNER_LOCK.acquire(blocking=False): - SCHEDULED_SCANNER_STATE["last_reason"] = "skipped_overlap" - return - - started = time.time() - - SCHEDULED_SCANNER_STATE["running"] = True - SCHEDULED_SCANNER_STATE["last_started_at"] = scheduled_now_iso() - SCHEDULED_SCANNER_STATE["last_reason"] = reason - SCHEDULED_SCANNER_STATE["last_errors"] = [] - - success_count = 0 - failure_count = 0 - errors = [] - - try: - miners = get_known_miners_for_scheduled_scan() - SCHEDULED_SCANNER_STATE["last_total_miners"] = len(miners) - - max_workers = max(1, int(SCHEDULED_SCANNER_STATE["max_workers"])) - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - future_map = { - executor.submit(refresh_one_known_miner_for_schedule, miner): miner - for miner in miners - } - - for future in as_completed(future_map): - miner = future_map[future] - - try: - future.result() - success_count += 1 - except Exception as exc: - failure_count += 1 - - if len(errors) < 25: - errors.append( - { - "ip_address": miner.get("ip_address"), - "mac": miner.get("current_mac"), - "room_name": miner.get("room_name"), - "error": str(exc), - } - ) - - SCHEDULED_SCANNER_STATE["last_success_count"] = success_count - SCHEDULED_SCANNER_STATE["last_failure_count"] = failure_count - SCHEDULED_SCANNER_STATE["last_errors"] = errors - SCHEDULED_SCANNER_STATE["total_runs"] += 1 - - except Exception as exc: - SCHEDULED_SCANNER_STATE["last_errors"] = [ - { - "error": str(exc), - "traceback": traceback.format_exc(limit=3), - } - ] - SCHEDULED_SCANNER_STATE["last_failure_count"] = failure_count - - finally: - finished = time.time() - - SCHEDULED_SCANNER_STATE["running"] = False - SCHEDULED_SCANNER_STATE["last_finished_at"] = scheduled_now_iso() - SCHEDULED_SCANNER_STATE["last_duration_seconds"] = round(finished - started, 2) - SCHEDULED_SCANNER_STATE["next_run_at"] = datetime.fromtimestamp( - finished + int(SCHEDULED_SCANNER_STATE["interval_seconds"]), - tz=timezone.utc, - ).isoformat() - - SCHEDULED_SCANNER_LOCK.release() + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) def scheduled_scanner_loop(): - interval = max(60, int(SCHEDULED_SCANNER_STATE["interval_seconds"])) - SCHEDULED_SCANNER_STATE["next_run_at"] = datetime.fromtimestamp( - time.time() + 10, - tz=timezone.utc, - ).isoformat() - - SCHEDULED_SCANNER_STOP.wait(10) - - while not SCHEDULED_SCANNER_STOP.is_set(): - if SCHEDULED_SCANNER_STATE["enabled"]: - run_scheduled_known_miner_scan(reason="scheduled") - - SCHEDULED_SCANNER_STOP.wait(interval) + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) @app.on_event("startup") def start_scheduled_scanner(): - if SCHEDULED_SCANNER_STATE["thread_started"]: - return - - if minerops_hub_mode_enabled(): - SCHEDULED_SCANNER_STATE["enabled"] = False - SCHEDULED_SCANNER_STATE["running"] = False - SCHEDULED_SCANNER_STATE["thread_started"] = False - SCHEDULED_SCANNER_STATE["last_reason"] = "hub_mode" - SCHEDULED_SCANNER_STATE["last_errors"] = [ - { - "message": "Hub Mode: local scanner disabled. Waiting for Windows site agent updates." - } - ] - SCHEDULED_SCANNER_STATE["next_run_at"] = None - return - - SCHEDULED_SCANNER_STATE["thread_started"] = True - - thread = threading.Thread( - target=scheduled_scanner_loop, - name="minerops-scheduled-scanner", - daemon=True, + # Retired in all modes: no background task may reach the legacy writer. + SCHEDULED_SCANNER_STATE.update( + enabled=False, running=False, thread_started=False, + last_reason="legacy_workflow_retired", next_run_at=None, + last_errors=[{"message": "Legacy scanner retired. Use authenticated Agent polling."}], ) - thread.start() @app.get("/api/scanner/status") def scanner_status(current_user: dict = Depends(get_current_user)): status = dict(SCHEDULED_SCANNER_STATE) - - if minerops_hub_mode_enabled(): - status["enabled"] = False - status["running"] = False - status["hub_mode"] = True - status["scanner_disabled"] = True - status["last_reason"] = "hub_mode" - status["last_errors"] = [ - { - "message": "Hub Mode: local scanner disabled. Waiting for Windows site agent updates." - } - ] - status["next_run_at"] = None - else: - status["hub_mode"] = False - status["scanner_disabled"] = False - + status.update( + enabled=False, running=False, scanner_disabled=True, + hub_mode=minerops_hub_mode_enabled(), + last_reason="legacy_workflow_retired", next_run_at=None, + last_errors=[{"message": "Legacy scanner retired. Use authenticated Agent polling."}], + ) return api_response(data=status) @app.post("/api/scanner/run-known-now") def scanner_run_known_now(current_user: dict = Depends(get_current_user)): - if minerops_hub_mode_enabled(): - status = dict(SCHEDULED_SCANNER_STATE) - status["enabled"] = False - status["running"] = False - status["hub_mode"] = True - status["scanner_disabled"] = True - status["last_reason"] = "hub_mode_manual_scan_blocked" - status["last_errors"] = [ - { - "message": "Hub Mode: manual local scan disabled. Use the Windows site agent to poll miners." - } - ] - status["next_run_at"] = None - - return api_response( - data=status, - meta={"message": "Hub Mode: local scanner disabled. Use the Windows site agent."}, - ) - - if SCHEDULED_SCANNER_STATE["running"]: - return api_response( - data=SCHEDULED_SCANNER_STATE, - meta={"message": "Scanner is already running."}, - ) - - thread = threading.Thread( - target=run_scheduled_known_miner_scan, - kwargs={"reason": "manual"}, - name="minerops-manual-known-scan", - daemon=True, - ) - thread.start() - - return api_response( - data=SCHEDULED_SCANNER_STATE, - meta={"message": "Known-miner scan started."}, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) # Patch: allow the scheduled scanner to call discover_one_miner when it expects FastAPI request=None. def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - signature = inspect.signature(func) - params = signature.parameters - - candidate_kwargs = { - "request": None, - "ip_address": ip_address, - "ip": ip_address, - "host": ip_address, - "room_id": room_id, - "source": "scheduled_known_miner_scan", - } - - kwargs = {} - - for name, param in params.items(): - if name in candidate_kwargs: - kwargs[name] = candidate_kwargs[name] - - missing_required = [] - - for name, param in params.items(): - if param.default is inspect._empty and name not in kwargs: - if param.kind in ( - inspect.Parameter.POSITIONAL_OR_KEYWORD, - inspect.Parameter.KEYWORD_ONLY, - ): - missing_required.append(name) - - if missing_required: - raise RuntimeError( - "discover_one_miner requires unsupported argument(s): " - + ", ".join(missing_required) - ) - - result = func(**kwargs) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) # Patch 2: scheduled scanner passes a body-like request object to discover_one_miner. from types import SimpleNamespace def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - signature = inspect.signature(func) - params = signature.parameters - - request_body = SimpleNamespace( - ip_address=ip_address, - room_id=room_id, - source="scheduled_known_miner_scan", - update_existing=True, - api_port=4028, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - candidate_kwargs = { - "request": request_body, - "body": request_body, - "payload": request_body, - "ip_address": ip_address, - "ip": ip_address, - "host": ip_address, - "room_id": room_id, - "source": "scheduled_known_miner_scan", - } - - kwargs = {} - - for name, param in params.items(): - if name in candidate_kwargs: - kwargs[name] = candidate_kwargs[name] - - missing_required = [] - - for name, param in params.items(): - if param.default is inspect._empty and name not in kwargs: - if param.kind in ( - inspect.Parameter.POSITIONAL_OR_KEYWORD, - inspect.Parameter.KEYWORD_ONLY, - ): - missing_required.append(name) - - if missing_required: - raise RuntimeError( - "discover_one_miner requires unsupported argument(s): " - + ", ".join(missing_required) - ) - - result = func(**kwargs) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result - # Patch 3: stronger fake request object for scheduled known-miner scans. class ScheduledDiscoverRequest: def __init__(self, ip_address: str, room_id: str | None = None): @@ -13997,61 +12747,12 @@ class ScheduledDiscoverRequest: def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - signature = inspect.signature(func) - params = signature.parameters - - request_body = ScheduledDiscoverRequest( - ip_address=ip_address, - room_id=room_id, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - candidate_kwargs = { - "request": request_body, - "body": request_body, - "payload": request_body, - "ip_address": ip_address, - "ip": ip_address, - "host": ip_address, - "port": 4028, - "api_port": 4028, - "room_id": room_id, - "source": "scheduled_known_miner_scan", - } - - kwargs = {} - - for name, param in params.items(): - if name in candidate_kwargs: - kwargs[name] = candidate_kwargs[name] - - missing_required = [] - - for name, param in params.items(): - if param.default is inspect._empty and name not in kwargs: - if param.kind in ( - inspect.Parameter.POSITIONAL_OR_KEYWORD, - inspect.Parameter.KEYWORD_ONLY, - ): - missing_required.append(name) - - if missing_required: - raise RuntimeError( - "discover_one_miner requires unsupported argument(s): " - + ", ".join(missing_required) - ) - - result = func(**kwargs) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result - # Patch 4: scheduled scanner request object with timeout_ms fields. class ScheduledDiscoverRequestV2: def __init__(self, ip_address: str, room_id: str | None = None): @@ -14107,140 +12808,20 @@ class ScheduledDiscoverRequestV2: def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - signature = inspect.signature(func) - params = signature.parameters - - request_body = ScheduledDiscoverRequestV2( - ip_address=ip_address, - room_id=room_id, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - candidate_kwargs = { - "request": request_body, - "body": request_body, - "payload": request_body, - "ip_address": ip_address, - "ip": ip_address, - "host": ip_address, - "port": 4028, - "api_port": 4028, - "room_id": room_id, - "source": "scheduled_known_miner_scan", - "timeout_ms": 5000, - "connect_timeout_ms": 3000, - "read_timeout_ms": 5000, - } - - kwargs = {} - - for name, param in params.items(): - if name in candidate_kwargs: - kwargs[name] = candidate_kwargs[name] - - missing_required = [] - - for name, param in params.items(): - if param.default is inspect._empty and name not in kwargs: - if param.kind in ( - inspect.Parameter.POSITIONAL_OR_KEYWORD, - inspect.Parameter.KEYWORD_ONLY, - ): - missing_required.append(name) - - if missing_required: - raise RuntimeError( - "discover_one_miner requires unsupported argument(s): " - + ", ".join(missing_required) - ) - - result = func(**kwargs) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result - # Patch 5: scheduled scanner passes an internal system user so FastAPI Depends is not used directly. def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - signature = inspect.signature(func) - params = signature.parameters - - request_body = ScheduledDiscoverRequestV2( - ip_address=ip_address, - room_id=room_id, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - system_user = { - "id": "00000000-0000-0000-0000-000000000000", - "user_id": "00000000-0000-0000-0000-000000000000", - "username": "system-scanner", - "role": "system", - "is_admin": True, - } - - candidate_kwargs = { - "request": request_body, - "body": request_body, - "payload": request_body, - - "ip_address": ip_address, - "ip": ip_address, - "host": ip_address, - - "port": 4028, - "api_port": 4028, - - "room_id": room_id, - "source": "scheduled_known_miner_scan", - - "timeout_ms": 5000, - "connect_timeout_ms": 3000, - "read_timeout_ms": 5000, - - "current_user": system_user, - "user": system_user, - "auth_user": system_user, - } - - kwargs = {} - - for name, param in params.items(): - if name in candidate_kwargs: - kwargs[name] = candidate_kwargs[name] - - missing_required = [] - - for name, param in params.items(): - if param.default is inspect._empty and name not in kwargs: - if param.kind in ( - inspect.Parameter.POSITIONAL_OR_KEYWORD, - inspect.Parameter.KEYWORD_ONLY, - ): - missing_required.append(name) - - if missing_required: - raise RuntimeError( - "discover_one_miner requires unsupported argument(s): " - + ", ".join(missing_required) - ) - - result = func(**kwargs) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result - # Patch 6: scheduled scanner uses a real user from the users table for audit logs. def get_scheduled_scanner_user(): conn = None @@ -14277,28 +12858,12 @@ def get_scheduled_scanner_user(): def call_discover_one_miner_flexible(ip_address: str, room_id: str | None = None): - func = globals().get("discover_one_miner") - - if func is None: - raise RuntimeError("discover_one_miner function was not found") - - request_body = ScheduledDiscoverRequestV2( - ip_address=ip_address, - room_id=room_id, + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - system_user = get_scheduled_scanner_user() - - result = func( - request=request_body, - current_user=system_user, - ) - - if inspect.iscoroutine(result): - result = asyncio.run(result) - - return result - @app.get("/api/dashboard/issues") def dashboard_issues( min_temp_c: float = 90, @@ -20475,240 +19040,23 @@ def miner_needs_legacy_summary_patch(miner_id: str): def apply_legacy_summary_to_miner(miner_id: str, ip_address: str, timeout_seconds: float = 1.5): - read_result = read_legacy_whatsminer_summary(ip_address, 4028, timeout_seconds) - - if not read_result.get("success"): - return { - "success": False, - "miner_id": miner_id, - "ip_address": ip_address, - "error": read_result.get("error"), - } - - summary = read_result.get("summary") or {} - - mhs_av = legacy_summary_to_float( - legacy_summary_get(summary, "MHS av", "MHS Av", "mhs_av", "MHS 5s", "HS RT") + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", ) - power_w = legacy_summary_to_float( - legacy_summary_get(summary, "Power", "power", "Power W") - ) - - power_rate = legacy_summary_to_float( - legacy_summary_get(summary, "Power Rate", "PowerRate", "power_rate") - ) - - chip_temp_max = legacy_summary_to_float( - legacy_summary_get(summary, "Chip Temp Max", "ChipTempMax", "chip_temp_max", "Temperature") - ) - - chip_temp_avg = legacy_summary_to_float( - legacy_summary_get(summary, "Chip Temp Avg", "ChipTempAvg", "chip_temp_avg", "Temperature") - ) - - fan_in = legacy_summary_to_int( - legacy_summary_get(summary, "Fan Speed In", "FanSpeedIn", "fan_speed_in") - ) - - fan_out = legacy_summary_to_int( - legacy_summary_get(summary, "Fan Speed Out", "FanSpeedOut", "fan_speed_out") - ) - - hash_stable = legacy_summary_to_bool( - legacy_summary_get(summary, "Hash Stable", "HashStable", "hash_stable") - ) - - pool_rejected_percent = legacy_summary_to_float( - legacy_summary_get(summary, "Pool Rejected%", "Pool Rejected", "pool_rejected_percent") - ) - - inferred_model = infer_legacy_whatsminer_model(summary) - - conn = None - - try: - conn = get_db_connection() - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - cur.execute( - """ - UPDATE miners - SET - status = 'online', - last_seen_at = now(), - current_ip = COALESCE(NULLIF(%s, '')::inet, current_ip), - model = COALESCE(NULLIF(model, ''), %s), - miner_type = COALESCE(NULLIF(miner_type, ''), %s) - WHERE id = %s::uuid - RETURNING - id::text, - current_mac, - host(current_ip) AS current_ip, - model, - miner_type; - """, - ( - ip_address, - inferred_model, - inferred_model, - miner_id, - ), - ) - - miner_row = cur.fetchone() - - if not miner_row: - raise HTTPException(status_code=404, detail="Miner not found") - - cur.execute( - """ - SELECT column_name - FROM information_schema.columns - WHERE table_name = 'miner_status_snapshots'; - """ - ) - - snapshot_columns = {row["column_name"] for row in cur.fetchall()} - - insert_columns = [] - insert_values_sql = [] - values = [] - - def add_value(column_name, value, sql_cast=None): - if column_name not in snapshot_columns: - return - - insert_columns.append(column_name) - - if sql_cast: - insert_values_sql.append(f"%s::{sql_cast}") - else: - insert_values_sql.append("%s") - - values.append(value) - - def add_now(column_name): - if column_name not in snapshot_columns: - return - - insert_columns.append(column_name) - insert_values_sql.append("now()") - - add_value("id", str(uuid4()), "uuid") - add_value("miner_id", miner_id, "uuid") - add_now("observed_at") - add_value("observed_ip", ip_address, "inet") - add_value("mhs_av", mhs_av) - add_value("power_w", power_w) - add_value("power_rate_w_per_th", power_rate) - add_value("chip_temp_max_c", chip_temp_max) - add_value("chip_temp_avg_c", chip_temp_avg) - add_value("fan_speed_in_rpm", fan_in) - add_value("fan_speed_out_rpm", fan_out) - add_value("hash_stable", hash_stable) - add_value("pool_rejected_percent", pool_rejected_percent) - add_now("created_at") - - if "raw_summary_json" in snapshot_columns: - import json - add_value("raw_summary_json", json.dumps(summary)) - - if len(insert_columns) > 0: - sql = f""" - INSERT INTO miner_status_snapshots ( - {", ".join(insert_columns)} - ) - VALUES ( - {", ".join(insert_values_sql)} - ); - """ - - cur.execute(sql, values) - - conn.commit() - - return { - "success": True, - "miner_id": miner_id, - "ip_address": ip_address, - "inferred_model": inferred_model, - "mhs_av": mhs_av, - "hashrate_ths": None if mhs_av is None else round(mhs_av / 1000000.0, 2), - "power_w": power_w, - "chip_temp_max_c": chip_temp_max, - "chip_temp_avg_c": chip_temp_avg, - "hash_stable": hash_stable, - "pool_rejected_percent": pool_rejected_percent, - } - - except Exception: - if conn is not None: - conn.rollback() - - raise - - finally: - if conn is not None: - conn.close() - @app.post("/api/tools/legacy-summary-refresh-one") def legacy_summary_refresh_one( request: LegacySummaryRefreshRequest, current_user: dict = Depends(get_current_user), ): - if not request.miner_id and not request.ip_address: - raise HTTPException(status_code=400, detail="Provide miner_id or ip_address") - - conn = None - - try: - conn = get_db_connection() - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - if request.miner_id: - cur.execute( - """ - SELECT id::text, host(current_ip) AS current_ip - FROM miners - WHERE id = %s::uuid; - """, - (request.miner_id,), - ) - else: - cur.execute( - """ - SELECT id::text, host(current_ip) AS current_ip - FROM miners - WHERE current_ip = %s::inet - LIMIT 1; - """, - (request.ip_address,), - ) - - miner = cur.fetchone() - - if not miner: - raise HTTPException(status_code=404, detail="Miner not found") - - miner_id = miner["id"] - ip_address = request.ip_address or miner["current_ip"] - - if not ip_address: - raise HTTPException(status_code=400, detail="Miner has no current IP address") - - result = apply_legacy_summary_to_miner( - miner_id, - ip_address, - request.timeout_seconds, - ) - - return api_response(data=result) - - finally: - if conn is not None: - conn.close() + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) @app.post("/api/tools/legacy-summary-refresh-missing") @@ -20716,127 +19064,30 @@ def legacy_summary_refresh_missing( request: LegacySummaryRefreshRequest, current_user: dict = Depends(get_current_user), ): - safe_limit = max(1, min(int(request.limit or 50), 300)) - - conn = None - - try: - conn = get_db_connection() - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - cur.execute( - """ - WITH latest AS ( - SELECT DISTINCT ON (miner_id) - miner_id, - mhs_av, - chip_temp_max_c, - chip_temp_avg_c, - observed_at - FROM miner_status_snapshots - ORDER BY miner_id, observed_at DESC - ) - SELECT - miners.id::text, - host(miners.current_ip) AS current_ip, - miners.current_mac, - miners.model, - miners.miner_type, - latest.mhs_av, - latest.chip_temp_max_c, - latest.chip_temp_avg_c - FROM miners - LEFT JOIN latest ON latest.miner_id = miners.id - WHERE miners.current_ip IS NOT NULL - AND ( - miners.model IS NULL - OR trim(miners.model) = '' - OR miners.miner_type IS NULL - OR trim(miners.miner_type) = '' - OR latest.mhs_av IS NULL - OR latest.chip_temp_max_c IS NULL - ) - ORDER BY miners.current_ip - LIMIT %s; - """, - (safe_limit,), - ) - - miners = [dict(row) for row in cur.fetchall()] - - results = [] - - for miner in miners: - result = apply_legacy_summary_to_miner( - miner["id"], - miner["current_ip"], - request.timeout_seconds, - ) - - results.append(result) - - return api_response( - data=results, - meta={ - "attempted": len(results), - "success": len([row for row in results if row.get("success")]), - "failed": len([row for row in results if not row.get("success")]), - }, - ) - - finally: - if conn is not None: - conn.close() + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) def legacy_find_miner_id_by_ip(ip_address: str): - conn = None - - try: - conn = get_db_connection() - - with conn.cursor(row_factory=psycopg.rows.dict_row) as cur: - cur.execute( - """ - SELECT id::text - FROM miners - WHERE current_ip = %s::inet - LIMIT 1; - """, - (ip_address,), - ) - - row = cur.fetchone() - - if not row: - return None - - return row["id"] - - finally: - if conn is not None: - conn.close() + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) try: _minerops_original_discover_one_miner_legacy_summary = discover_one_miner def discover_one_miner(request, current_user: dict = Depends(get_current_user)): - result = _minerops_original_discover_one_miner_legacy_summary(request, current_user) - - try: - ip_address = getattr(request, "ip_address", None) or getattr(request, "ip", None) - - if ip_address: - miner_id = legacy_find_miner_id_by_ip(ip_address) - - if miner_id and miner_needs_legacy_summary_patch(miner_id): - apply_legacy_summary_to_miner(miner_id, ip_address, 1.5) - - except Exception as legacy_error: - print(f"Legacy summary fallback failed: {legacy_error}") - - return result + raise HTTPException( + status_code=410, + detail="Legacy inventory discovery and summary writes are retired. " + "Use authenticated Agent observations, discovery review, and inventory-bound Agent polling.", + ) except NameError: pass @@ -23822,68 +22073,7 @@ async def minerops_agent_miner_snapshots( authenticated_agent_id: str = Depends(require_agent_identity), zone_service: AgentZoneService = Depends(get_agent_zone_service), ): - import json - - await ensure_minerops_agent_tables() - - try: - body = await request.json() - except Exception: - body = {} - - zone_service.validate_agent_identity( - authenticated_agent_id, str(body.get("agent_id") or "").strip() or None - ) - site_name = str(body.get("site_name") or "Unassigned").strip() - miners = body.get("miners") - if not isinstance(miners, list): - miners = [] - - miner_ids = [] - for item in miners: - if not isinstance(item, dict): - continue - try: - miner_ids.append(UUID(str(item.get("miner_id") or ""))) - except ValueError: - continue - ownership = zone_service.associate_miners_from_agent( - authenticated_agent_id, miner_ids - ) - - conn = await minerops_command_db_connect() - try: - row = await conn.fetchrow( - """ - INSERT INTO minerops_agent_snapshots ( - agent_id, agent_zone_id, site_name, snapshot_kind, - miner_count, payload - ) - VALUES ($1, $2::uuid, $3, 'miner_poll', $4, $5::jsonb) - RETURNING id, received_at - """, - authenticated_agent_id, - str(ownership.agent_zone_id) if ownership.agent_zone_id else None, - site_name, - len(miners), - json.dumps(body), - ) - finally: - await conn.close() - - apply_result = await minerops_apply_agent_snapshot_to_live_tables( - body, ownership - ) - return { - "ok": True, - "snapshot_id": row["id"] if row else None, - "received_at": row["received_at"].isoformat() if row and row["received_at"] else None, - "miner_count": len(miners), - "agent_zone_id": str(ownership.agent_zone_id) if ownership.agent_zone_id else None, - "routing_reason": ownership.routing_reason, - "apply_result": apply_result, - "message": "Miner snapshot received, staged, and applied to live tables.", - } + return await discovery_dispatch('snapshots', request, agent_id=authenticated_agent_id) @app.get("/api/agent/commands/pending-disabled-old") @@ -24047,59 +22237,7 @@ async def minerops_agent_known_miners( authenticated_agent_id: str = Depends(require_agent_identity), zone_service: AgentZoneService = Depends(get_agent_zone_service), ): - routing = zone_service.routing_context(authenticated_agent_id) - if routing.routing_reason or routing.agent_zone_id is None: - raise HTTPException( - status_code=409, - detail=( - "Agent routing unavailable: " - + (routing.routing_reason or "agent_zone_unassigned") - ), - ) - - conn = await minerops_command_db_connect() - - try: - rows = await conn.fetch( - """ - SELECT - miners.id::text AS miner_id, - host(miners.current_ip) AS ip_address, - miners.current_mac AS mac_address, - COALESCE(rooms.name, 'Unassigned') AS room_name, - COALESCE(NULLIF(miners.miner_type, ''), NULLIF(miners.model, ''), '') AS model, - COALESCE(miners.status, '') AS status - FROM miners - LEFT JOIN rooms ON rooms.id = miners.room_id - WHERE miners.current_ip IS NOT NULL - AND miners.agent_zone_id = $1::uuid - ORDER BY rooms.sort_order NULLS LAST, rooms.name NULLS LAST, miners.current_ip - """, - str(routing.agent_zone_id), - ) - finally: - await conn.close() - - miners = [] - - for row in rows: - miners.append( - { - "miner_id": row["miner_id"], - "ip_address": row["ip_address"], - "mac_address": row["mac_address"], - "room_name": row["room_name"], - "model": row["model"], - "status": row["status"], - } - ) - - return { - "ok": True, - "count": len(miners), - "miners": miners, - "message": "Known miners for site agent polling.", - } + return await discovery_dispatch('known', agent_id=authenticated_agent_id) @@ -24184,150 +22322,7 @@ def minerops_agent_parse_summary(miner_payload: dict) -> dict: async def minerops_apply_agent_snapshot_to_live_tables( snapshot_body: dict, ownership: MinerZoneAssociationResult ) -> dict: - import json - import uuid - from datetime import datetime, timezone - - miners = snapshot_body.get("miners") - if not isinstance(miners, list): - miners = [] - - now = datetime.now(timezone.utc) - applied = 0 - online_count = 0 - offline_count = 0 - errors = [] - conn = await minerops_command_db_connect() - - try: - for item in miners: - if not isinstance(item, dict): - continue - - miner_id = str(item.get("miner_id") or "").strip() - ip_address = str(item.get("ip_address") or "").strip() - online = bool(item.get("online")) - if not miner_id: - continue - try: - miner_uuid = UUID(miner_id) - except ValueError: - errors.append({"miner_id": miner_id, "error": "invalid_miner_id"}) - continue - - if ownership.routing_reason: - errors.append( - {"miner_id": miner_id, "error": ownership.routing_reason} - ) - continue - zone_outcome = ownership.outcomes.get(miner_uuid, "miner_not_found") - if zone_outcome not in {"assigned", "preserved"}: - errors.append({"miner_id": miner_id, "error": zone_outcome}) - continue - - parsed = minerops_agent_parse_summary(item) - raw_summary = parsed["raw"] - msg = parsed["msg"] - status = "online" if online else "offline" - mhs_av = minerops_agent_to_float(msg.get("MHS av")) - mhs_1m = minerops_agent_to_float(msg.get("MHS 1m")) - mhs_15m = minerops_agent_to_float(msg.get("MHS 15m")) - hs_rt = minerops_agent_to_float(msg.get("HS RT")) - power_w = minerops_agent_to_float(msg.get("Power")) - power_rate = minerops_agent_to_float(msg.get("Power Rate")) - env_temp = minerops_agent_to_float(msg.get("Env Temp")) - chip_temp_min = minerops_agent_to_float(msg.get("Chip Temp Min")) - chip_temp_max = minerops_agent_to_float(msg.get("Chip Temp Max")) - chip_temp_avg = minerops_agent_to_float(msg.get("Chip Temp Avg")) - fan_in = minerops_agent_to_int(msg.get("Fan Speed In")) - fan_out = minerops_agent_to_int(msg.get("Fan Speed Out")) - uptime = minerops_agent_to_int(msg.get("Uptime")) - hash_stable = minerops_agent_to_bool(msg.get("Hash Stable")) - pool_rejected = minerops_agent_to_float(msg.get("Pool Rejected%")) - - try: - if online: - online_count += 1 - await conn.execute( - """ - UPDATE miners - SET status = $2, - current_ip = COALESCE(NULLIF($3, '')::inet, current_ip), - last_seen_at = $4, - updated_at = now() - WHERE id = $1::uuid - """, - miner_id, - status, - ip_address, - now, - ) - await conn.execute( - """ - INSERT INTO miner_status_snapshots ( - id, miner_id, observed_ip, observed_at, mhs_av, - mhs_1m, mhs_15m, hs_rt, power_w, - power_rate_w_per_th, env_temp_c, chip_temp_min_c, - chip_temp_max_c, chip_temp_avg_c, fan_speed_in_rpm, - fan_speed_out_rpm, uptime_seconds, hash_stable, - pool_rejected_percent, raw_summary_json, created_at - ) - VALUES ( - $1::uuid, $2::uuid, NULLIF($3, '')::inet, $4, - $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, - $15, $16, $17, $18, $19, $20::jsonb, now() - ) - """, - str(uuid.uuid4()), - miner_id, - ip_address, - now, - mhs_av, - mhs_1m, - mhs_15m, - hs_rt, - power_w, - power_rate, - env_temp, - chip_temp_min, - chip_temp_max, - chip_temp_avg, - fan_in, - fan_out, - uptime, - hash_stable, - pool_rejected, - json.dumps(raw_summary), - ) - else: - offline_count += 1 - await conn.execute( - """ - UPDATE miners - SET status = $2, - current_ip = COALESCE(NULLIF($3, '')::inet, current_ip), - updated_at = now() - WHERE id = $1::uuid - """, - miner_id, - status, - ip_address, - ) - applied += 1 - except Exception as exc: - if len(errors) < 20: - errors.append( - {"miner_id": miner_id, "ip_address": ip_address, "error": str(exc)} - ) - finally: - await conn.close() - - return { - "applied": applied, - "online": online_count, - "offline": offline_count, - "errors": errors, - } + raise HTTPException(409, "discovery_versioned_snapshot_required") # ----------------------------- @@ -24373,58 +22368,8 @@ async def startup_ensure_minerops_agent_command_queue_table(): @app.post("/api/agent/commands/run-poll-now") -async def minerops_queue_agent_poll_now(current_user: dict = Depends(get_current_user)): - import uuid - import json - - await minerops_ensure_agent_command_queue_table() - - command_id = str(uuid.uuid4()) - username = str(current_user.get("username") or current_user.get("email") or "unknown") - user_id = str(current_user.get("id") or "") - - conn = await minerops_command_db_connect() - - try: - await conn.execute( - """ - INSERT INTO minerops_agent_command_queue ( - id, - agent_id, - command_type, - command_status, - requested_by_user_id, - requested_by_username, - request_payload, - created_at - ) - VALUES ( - $1, - $2, - $3, - 'pending', - $4, - $5, - $6::jsonb, - now() - ) - """, - command_id, - "plexus-windows-agent", - "poll_known_miners_now", - user_id, - username, - json.dumps({"source": "web", "reason": "manual_poll_now"}), - ) - finally: - await conn.close() - - return { - "ok": True, - "command_id": command_id, - "command_type": "poll_known_miners_now", - "message": "Agent poll command queued.", - } +async def minerops_queue_agent_poll_now(request: Request, current_user: dict = Depends(get_current_user)): + return await discovery_dispatch('queue', request, user=current_user, kind='poll_known_miners_now') @app.get("/api/agent/commands/status") @@ -24754,6 +22699,10 @@ async def minerops_agent_control_page():