634 lines
19 KiB
Python
634 lines
19 KiB
Python
"""Runtime PostgreSQL isolation tests for Agent-owned machine transport."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import unittest
|
|
from uuid import UUID
|
|
|
|
import psycopg
|
|
from fastapi import HTTPException
|
|
from starlette.requests import Request
|
|
|
|
from app.agent_zones.repository import AgentZoneRepository
|
|
from app.agent_zones.service import AgentZoneService
|
|
from app.config import settings
|
|
from app.main import (
|
|
minerops_agent_api3_enrichment,
|
|
minerops_agent_miner_enrichment_v2,
|
|
minerops_agent_miner_enrichment_v3,
|
|
)
|
|
|
|
|
|
SITE_A = UUID("00000000-0000-4000-8000-000000001710")
|
|
SITE_B = UUID("00000000-0000-4000-8000-000000001711")
|
|
ZONE_A = UUID("00000000-0000-4000-8000-000000001712")
|
|
ZONE_B = UUID("00000000-0000-4000-8000-000000001713")
|
|
MINER_A = UUID("00000000-0000-4000-8000-000000001714")
|
|
MINER_B = UUID("00000000-0000-4000-8000-000000001715")
|
|
|
|
AGENT_A = "agent-transport-isolation-a"
|
|
AGENT_B = "agent-transport-isolation-b"
|
|
|
|
IP_A = "198.51.100.171"
|
|
IP_B = "198.51.100.172"
|
|
|
|
MAC_A = "02:00:00:00:17:A1"
|
|
MAC_B = "02:00:00:00:17:B1"
|
|
|
|
|
|
def connect():
|
|
return psycopg.connect(
|
|
host=settings.postgres_host,
|
|
port=settings.postgres_port,
|
|
dbname=settings.postgres_db,
|
|
user=settings.postgres_user,
|
|
password=settings.postgres_password,
|
|
connect_timeout=5,
|
|
)
|
|
|
|
|
|
def json_request(path, payload):
|
|
body = json.dumps(payload).encode("utf-8")
|
|
delivered = False
|
|
|
|
async def receive():
|
|
nonlocal delivered
|
|
|
|
if delivered:
|
|
return {
|
|
"type": "http.request",
|
|
"body": b"",
|
|
"more_body": False,
|
|
}
|
|
|
|
delivered = True
|
|
|
|
return {
|
|
"type": "http.request",
|
|
"body": body,
|
|
"more_body": False,
|
|
}
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"asgi": {"version": "3.0"},
|
|
"http_version": "1.1",
|
|
"method": "POST",
|
|
"scheme": "http",
|
|
"path": path,
|
|
"raw_path": path.encode("ascii"),
|
|
"query_string": b"",
|
|
"headers": [(b"content-type", b"application/json")],
|
|
"client": ("testclient", 12345),
|
|
"server": ("testserver", 80),
|
|
}
|
|
|
|
return Request(scope, receive)
|
|
|
|
|
|
class AgentTransportIsolationPostgresTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if os.environ.get(
|
|
"MINEROPS_TRANSPORT_ISOLATION_DISPOSABLE"
|
|
) != "1":
|
|
raise unittest.SkipTest(
|
|
"Runtime Agent transport isolation is executed "
|
|
"by its dedicated disposable PostgreSQL runner."
|
|
)
|
|
|
|
connection = connect()
|
|
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO sites (id, name)
|
|
VALUES
|
|
(%s, 'Transport Isolation Site A'),
|
|
(%s, 'Transport Isolation Site B')
|
|
ON CONFLICT (id) DO NOTHING
|
|
""",
|
|
(SITE_A, SITE_B),
|
|
)
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO agent_zones (
|
|
id,
|
|
site_id,
|
|
name,
|
|
is_enabled
|
|
)
|
|
VALUES
|
|
(
|
|
%s,
|
|
%s,
|
|
'Transport Isolation Zone A',
|
|
true
|
|
),
|
|
(
|
|
%s,
|
|
%s,
|
|
'Transport Isolation Zone B',
|
|
true
|
|
)
|
|
ON CONFLICT (id) DO UPDATE
|
|
SET is_enabled = true
|
|
""",
|
|
(ZONE_A, SITE_A, ZONE_B, SITE_B),
|
|
)
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO minerops_site_agents (
|
|
agent_id,
|
|
site_id,
|
|
agent_zone_id,
|
|
site_name,
|
|
last_payload
|
|
)
|
|
VALUES
|
|
(
|
|
%s,
|
|
%s,
|
|
%s,
|
|
'Transport Isolation Site A',
|
|
'{}'::jsonb
|
|
),
|
|
(
|
|
%s,
|
|
%s,
|
|
%s,
|
|
'Transport Isolation Site B',
|
|
'{}'::jsonb
|
|
)
|
|
ON CONFLICT (agent_id) DO UPDATE
|
|
SET
|
|
site_id = EXCLUDED.site_id,
|
|
agent_zone_id = EXCLUDED.agent_zone_id,
|
|
site_name = EXCLUDED.site_name,
|
|
last_payload = EXCLUDED.last_payload
|
|
""",
|
|
(
|
|
AGENT_A,
|
|
SITE_A,
|
|
ZONE_A,
|
|
AGENT_B,
|
|
SITE_B,
|
|
ZONE_B,
|
|
),
|
|
)
|
|
|
|
connection.commit()
|
|
|
|
finally:
|
|
connection.close()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
# Immutable identity/history rows intentionally are not deleted.
|
|
# The disposable Compose runner destroys the database volume.
|
|
pass
|
|
|
|
@staticmethod
|
|
def _delete_optional_rows(cur, table, predicate, params):
|
|
cur.execute(
|
|
"SELECT to_regclass(%s)",
|
|
(f"public.{table}",),
|
|
)
|
|
|
|
if cur.fetchone()[0] is not None:
|
|
cur.execute(
|
|
f"DELETE FROM {table} WHERE {predicate}",
|
|
params,
|
|
)
|
|
|
|
def setUp(self):
|
|
self.zone_service = AgentZoneService(
|
|
AgentZoneRepository(connect)
|
|
)
|
|
|
|
connection = connect()
|
|
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miners (
|
|
id,
|
|
site_id,
|
|
agent_zone_id,
|
|
asset_tag,
|
|
current_mac,
|
|
current_ip,
|
|
hostname,
|
|
miner_sn,
|
|
model,
|
|
miner_type,
|
|
api_version,
|
|
firmware_version,
|
|
platform,
|
|
status,
|
|
notes,
|
|
api3_supported
|
|
)
|
|
VALUES
|
|
(
|
|
%s,
|
|
%s,
|
|
%s,
|
|
'TRANSPORT-ZONE-A',
|
|
%s,
|
|
%s::inet,
|
|
'zone-a-before',
|
|
'ZONE-A-SERIAL',
|
|
'Zone A Model',
|
|
'Zone A Model',
|
|
'1.0-a',
|
|
'fw-a',
|
|
'platform-a',
|
|
'online',
|
|
'Zone A baseline',
|
|
false
|
|
),
|
|
(
|
|
%s,
|
|
%s,
|
|
%s,
|
|
'TRANSPORT-ZONE-B',
|
|
%s,
|
|
%s::inet,
|
|
'zone-b-before',
|
|
'ZONE-B-SERIAL',
|
|
'Zone B Model',
|
|
'Zone B Model',
|
|
'1.0-b',
|
|
'fw-b',
|
|
'platform-b',
|
|
'online',
|
|
'Zone B baseline',
|
|
false
|
|
)
|
|
ON CONFLICT (id) DO NOTHING
|
|
""",
|
|
(
|
|
MINER_A,
|
|
SITE_A,
|
|
ZONE_A,
|
|
MAC_A,
|
|
IP_A,
|
|
MINER_B,
|
|
SITE_B,
|
|
ZONE_B,
|
|
MAC_B,
|
|
IP_B,
|
|
),
|
|
)
|
|
|
|
connection.commit()
|
|
|
|
finally:
|
|
connection.close()
|
|
|
|
def _table_exists(self, cur, table):
|
|
cur.execute(
|
|
"SELECT to_regclass(%s)",
|
|
(f"public.{table}",),
|
|
)
|
|
return cur.fetchone()[0] is not None
|
|
|
|
def _snapshot_rows(self, cur, table):
|
|
cur.execute(
|
|
f"""
|
|
SELECT to_jsonb(t)
|
|
FROM {table} AS t
|
|
WHERE miner_id IN (%s, %s)
|
|
ORDER BY to_jsonb(t)::text
|
|
""",
|
|
(MINER_A, MINER_B),
|
|
)
|
|
|
|
return [row[0] for row in cur.fetchall()]
|
|
|
|
def _count_optional_miner_rows(self, cur, table, miner_id):
|
|
if not self._table_exists(cur, table):
|
|
return 0
|
|
|
|
cur.execute(
|
|
f"""
|
|
SELECT COUNT(*)
|
|
FROM {table}
|
|
WHERE miner_id = %s
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
|
|
return cur.fetchone()[0]
|
|
|
|
def _capture_inventory_state(self):
|
|
connection = connect()
|
|
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id::text, to_jsonb(m)
|
|
FROM miners AS m
|
|
WHERE id IN (%s, %s)
|
|
ORDER BY id
|
|
""",
|
|
(MINER_A, MINER_B),
|
|
)
|
|
|
|
miners = {
|
|
row[0]: row[1]
|
|
for row in cur.fetchall()
|
|
}
|
|
|
|
ownerships = self._snapshot_rows(
|
|
cur,
|
|
"miner_mac_ownerships",
|
|
)
|
|
|
|
identities = self._snapshot_rows(
|
|
cur,
|
|
"miner_network_identities",
|
|
)
|
|
|
|
counts = {}
|
|
|
|
for table in (
|
|
"miner_status_snapshots",
|
|
"miner_api3_enrichment_snapshots",
|
|
"miner_api3_hashboard_snapshots",
|
|
"miner_pool_snapshots",
|
|
):
|
|
counts[(table, "A")] = (
|
|
self._count_optional_miner_rows(
|
|
cur,
|
|
table,
|
|
MINER_A,
|
|
)
|
|
)
|
|
counts[(table, "B")] = (
|
|
self._count_optional_miner_rows(
|
|
cur,
|
|
table,
|
|
MINER_B,
|
|
)
|
|
)
|
|
|
|
return {
|
|
"miners": miners,
|
|
"mac_ownerships": ownerships,
|
|
"network_identities": identities,
|
|
"snapshot_counts": counts,
|
|
}
|
|
|
|
finally:
|
|
connection.close()
|
|
|
|
def _cross_zone_api3_probe_count(self):
|
|
connection = connect()
|
|
|
|
try:
|
|
with connection.cursor() as cur:
|
|
if not self._table_exists(
|
|
cur,
|
|
"miner_api3_enrichment_snapshots",
|
|
):
|
|
return 0
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM miner_api3_enrichment_snapshots
|
|
WHERE agent_id = %s
|
|
AND (
|
|
host(ip_address) = %s
|
|
OR upper(mac_address) = upper(%s)
|
|
)
|
|
""",
|
|
(AGENT_A, IP_B, MAC_B),
|
|
)
|
|
|
|
return cur.fetchone()[0]
|
|
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_v2_cannot_address_other_zone_miner_directly(self):
|
|
before = self._capture_inventory_state()
|
|
|
|
request = json_request(
|
|
"/api/agent/miner-enrichment-v2",
|
|
{
|
|
"miner_id": str(MINER_B),
|
|
"mac_address": MAC_B,
|
|
"hostname": "zone-a-must-not-write-this",
|
|
"model": "cross-zone-v2-probe",
|
|
},
|
|
)
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
asyncio.run(
|
|
minerops_agent_miner_enrichment_v2(
|
|
request,
|
|
AGENT_A,
|
|
self.zone_service,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 404)
|
|
|
|
after = self._capture_inventory_state()
|
|
|
|
self.assertEqual(after, before)
|
|
|
|
def test_v2_cross_zone_mac_merge_cannot_mutate_either_miner(self):
|
|
before = self._capture_inventory_state()
|
|
|
|
request = json_request(
|
|
"/api/agent/miner-enrichment-v2",
|
|
{
|
|
"miner_id": str(MINER_A),
|
|
"ip_address": IP_A,
|
|
"mac_address": MAC_B,
|
|
"hostname": "cross-zone-merge-probe",
|
|
"miner_sn": "MUST-NOT-CROSS-ZONES",
|
|
"model": "Cross Zone Merge Probe",
|
|
},
|
|
)
|
|
|
|
error = None
|
|
|
|
try:
|
|
asyncio.run(
|
|
minerops_agent_miner_enrichment_v2(
|
|
request,
|
|
AGENT_A,
|
|
self.zone_service,
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
error = exc
|
|
|
|
self.assertIsNotNone(
|
|
error,
|
|
"A Zone A miner must not successfully claim a Zone B MAC",
|
|
)
|
|
|
|
after = self._capture_inventory_state()
|
|
|
|
self.assertEqual(after, before)
|
|
|
|
def test_v3_cannot_match_or_mutate_other_zone_miner(self):
|
|
before = self._capture_inventory_state()
|
|
|
|
request = json_request(
|
|
"/api/agent/miner-enrichment-v3",
|
|
{
|
|
"miner_id": str(MINER_B),
|
|
"ip_address": IP_B,
|
|
"mac_address": MAC_B,
|
|
"hostname": "zone-a-v3-cross-zone-probe",
|
|
"miner_sn": "V3-MUST-NOT-MUTATE",
|
|
"model": "V3 Cross Zone Probe",
|
|
"api_version": "99.99",
|
|
"firmware_version": "cross-zone",
|
|
"platform": "cross-zone",
|
|
"raw_version": {"cross_zone": True},
|
|
"raw_miner_info": {"cross_zone": True},
|
|
},
|
|
)
|
|
|
|
result = asyncio.run(
|
|
minerops_agent_miner_enrichment_v3(
|
|
request,
|
|
AGENT_A,
|
|
self.zone_service,
|
|
)
|
|
)
|
|
|
|
self.assertFalse(result["matched"])
|
|
|
|
after = self._capture_inventory_state()
|
|
|
|
self.assertEqual(after, before)
|
|
|
|
def test_api3_ingestion_cannot_associate_or_mutate_other_zone_miner(self):
|
|
before = self._capture_inventory_state()
|
|
|
|
self.assertEqual(
|
|
self._cross_zone_api3_probe_count(),
|
|
0,
|
|
)
|
|
|
|
request = json_request(
|
|
"/api/agent/api3-enrichment",
|
|
{
|
|
"agent_id": AGENT_B,
|
|
"site_name": "spoofed-zone-b",
|
|
"items": [
|
|
{
|
|
"ip": IP_B,
|
|
"mac": MAC_B,
|
|
"api3_supported": True,
|
|
"api3_error": None,
|
|
"device_info": {
|
|
"msg": {
|
|
"network": {
|
|
"ip": IP_B,
|
|
"mac": MAC_B,
|
|
},
|
|
"miner": {
|
|
"type": "Cross Zone API3 Probe",
|
|
"miner-sn": "API3-MUST-NOT-MUTATE",
|
|
},
|
|
"system": {
|
|
"api": "99.99",
|
|
"apiswitch": "on",
|
|
"fwversion": "cross-zone-fw",
|
|
"platform": "cross-zone-platform",
|
|
},
|
|
"power": {
|
|
"sn": "cross-zone-power",
|
|
"model": "cross-zone-psu",
|
|
},
|
|
}
|
|
},
|
|
"miner_setting": {
|
|
"msg": {
|
|
"power-mode": "high",
|
|
}
|
|
},
|
|
"miner_status": {
|
|
"msg": {
|
|
"summary": {
|
|
"hash-average": 999,
|
|
"power-realtime": 9999,
|
|
"power-rate": 99,
|
|
"chip-temp-avg": 99,
|
|
"chip-temp-max": 100,
|
|
},
|
|
"pools": [
|
|
{
|
|
"id": 0,
|
|
"status": "Alive",
|
|
"account": "cross-zone-worker",
|
|
"url": "stratum+tcp://invalid.example:3333",
|
|
}
|
|
],
|
|
"edevs": [
|
|
{
|
|
"slot": 0,
|
|
"hash-average": 999,
|
|
"factory-hash": 999,
|
|
"freq": 999,
|
|
"effective-chips": 999,
|
|
}
|
|
],
|
|
}
|
|
},
|
|
"fan_setting": {
|
|
"msg": {
|
|
"mode": "cross-zone",
|
|
}
|
|
},
|
|
"custom_data": {
|
|
"msg": {
|
|
"CustomerSn": "CROSS-ZONE-CUSTOMER",
|
|
}
|
|
},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
asyncio.run(
|
|
minerops_agent_api3_enrichment(
|
|
request,
|
|
AGENT_A,
|
|
self.zone_service,
|
|
)
|
|
)
|
|
|
|
after = self._capture_inventory_state()
|
|
|
|
self.assertEqual(after, before)
|
|
|
|
self.assertEqual(
|
|
self._cross_zone_api3_probe_count(),
|
|
0,
|
|
(
|
|
"Agent A must not persist API3 enrichment associated "
|
|
"with Zone B identity data"
|
|
),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|