"""Audit F1-F3 reversals: real registered HTTP/transactions, no physical devices.""" import ast from concurrent.futures import ThreadPoolExecutor import os from pathlib import Path import threading import time import unittest from unittest.mock import patch from uuid import uuid4 from fastapi import HTTPException import app.main as main from app.identity import create_access_token from tests.test_discovery_security_postgres import DiscoveryFixtures, connect class LegacyInventorySecurityTests(DiscoveryFixtures, unittest.TestCase): def upload(self, text, query='', headers=None): return self.client.post( '/api/imports/whatsminer-tool-scan?dry_run=false&update_existing=true' + query, headers=self.admin if headers is None else headers, files={'file': ('simulated.csv', text.encode(), 'text/csv')}, ) def denied(self, response, before, status): self.assertEqual(response.status_code, status, response.text) self.assertEqual(before, self.durable()) self.transport.assert_not_called() def test_discovery_audit_reproductions_all_rejected_without_reads(self): self.miner(1, ip='198.51.100.111') self.miner(ip='198.51.100.125') self.miner(1, ip='198.51.100.125') for mode in ['true', 'false']: for ip in ['198.51.100.111', '198.51.100.125', '203.0.113.209']: with self.subTest(mode=mode, ip=ip), patch.dict(os.environ, MINEROPS_HUB_MODE=mode): before = self.durable() response = self.client.post('/api/miners/discover-one', headers=self.admin, json={'ip_address': ip}) self.denied(response, before, 410) def test_missing_disabled_agent_and_zone_cannot_restore_legacy_discovery(self): with connect() as c: c.execute('UPDATE agent_zones SET is_enabled=false WHERE id=%s', (self.zones[0],)) c.execute('UPDATE minerops_site_agents SET credential_revoked_at=now() WHERE agent_id=%s', (self.agents[0],)) c.execute('DELETE FROM minerops_site_agents WHERE agent_id=%s', (self.agents[1],)) before = self.durable() r = self.client.post('/api/miners/discover-one', headers=self.admin, json={'ip_address': '198.51.100.111'}) self.denied(r, before, 410) def test_inventory_only_permission_cannot_replace_mac(self): role, uid = uuid4(), uuid4() with connect() as c: c.execute('INSERT INTO roles(id,name) VALUES (%s,%s)', (role, str(role))) c.execute("INSERT INTO role_permissions(role_id,permission_id) SELECT %s,id FROM permissions WHERE key='miners.update'", (role,)) c.execute("INSERT INTO users(id,username,display_name,password_hash,role_id) VALUES (%s,%s,'Inventory only','unused',%s)", (uid, str(uid), role)) headers = {'Authorization': 'Bearer ' + create_access_token(str(uid), str(uid), str(role))} mid, _ = self.miner(ip='198.51.100.127') before = self.durable() for path, body, status in [ ('/api/agent/discoveries/review', {'discovery_id': str(uuid4()), 'version': 'x', 'action': 'ignore'}, 403), (f'/api/miners/{mid}/replace-control-board', {}, 403), ('/api/miners/discover-one', {'ip_address': '198.51.100.127'}, 410), ]: self.denied(self.client.post(path, headers=headers, json=body), before, status) def test_range_and_manual_scheduler_retired_in_both_modes(self): for mode in ['true', 'false']: with patch.dict(os.environ, MINEROPS_HUB_MODE=mode): for path in [f'/api/scan-ranges/{uuid4()}/discover-now', '/api/scanner/run-known-now']: before = self.durable() self.denied(self.client.post(path, headers=self.admin, json={}), before, 410) with patch('app.main.threading.Thread') as thread: main.start_scheduled_scanner() thread.assert_not_called() status = self.client.get('/api/scanner/status', headers=self.admin).json()['data'] self.assertFalse(status['enabled']) self.assertTrue(status['scanner_disabled']) def test_all_legacy_helper_definitions_and_saved_aliases_are_retired(self): # Execute every original definition, including shadowed functions; decorators # are omitted so this cannot mutate the application's registered routes. tree = ast.parse(Path(main.__file__).read_text()) names = {'find_existing_miner', 'discover_one_miner', 'discover_scan_range_now', 'call_discover_one_miner_flexible', 'refresh_one_known_miner_for_schedule', 'run_scheduled_known_miner_scan', 'scheduled_scanner_loop', 'apply_legacy_summary_to_miner', 'legacy_find_miner_id_by_ip'} import inspect for node in ast.walk(tree): if isinstance(node, ast.FunctionDef) and node.name in names: with self.subTest(helper=node.name, line=node.lineno): node.decorator_list = [] namespace = dict(vars(main)) exec(compile(ast.Module(body=[node], type_ignores=[]), '', 'exec'), namespace) fn = namespace[node.name] kwargs = {name: None for name, p in inspect.signature(fn).parameters.items() if p.default is inspect.Parameter.empty} with patch('app.main.get_db_connection', side_effect=AssertionError('No legacy database access')): with self.assertRaises(HTTPException) as exc: fn(**kwargs) self.assertEqual(exc.exception.status_code, 410) with self.assertRaises(HTTPException) as exc: main._minerops_original_discover_one_miner_legacy_summary(None) self.assertEqual(exc.exception.status_code, 410) def test_summary_id_ip_disagreement_and_bulk_retired(self): a, _ = self.miner(ip='198.51.100.121') self.miner(1, ip='198.51.100.122') with patch('app.main.read_legacy_whatsminer_summary', side_effect=AssertionError('No summary transport')) as read: for mode in ['true', 'false']: with patch.dict(os.environ, MINEROPS_HUB_MODE=mode): for path, body in [ ('/api/tools/legacy-summary-refresh-one', {'miner_id': str(a), 'ip_address': '198.51.100.122'}), ('/api/tools/legacy-summary-refresh-one', {'miner_id': str(a)}), ('/api/tools/legacy-summary-refresh-missing', {}), ]: before = self.durable() self.denied(self.client.post(path, headers=self.admin, json=body), before, 410) read.assert_not_called() def test_summary_cannot_start_inflight_read_or_overwrite_newer_owner(self): mid, _ = self.miner(ip='198.51.100.123') with patch('app.main.read_legacy_whatsminer_summary') as read, ThreadPoolExecutor(2) as pool: future = pool.submit(lambda: self.client.post('/api/tools/legacy-summary-refresh-one', headers=self.admin, json={'miner_id': str(mid)})) with connect() as c: c.execute("UPDATE miners SET site_id=%s,agent_zone_id=%s,current_ip='198.51.100.124',updated_at=clock_timestamp() WHERE id=%s", (self.sites[1], self.zones[1], mid)) self.assertEqual(future.result(timeout=8).status_code, 410) before = self.durable() with self.assertRaises(HTTPException): main.apply_legacy_summary_to_miner(str(mid), '198.51.100.123') self.assertEqual(before, self.durable()) self.assertEqual(self.scalar('SELECT host(current_ip) FROM miners WHERE id=%s', (mid,)), '198.51.100.124') read.assert_not_called() def test_legacy_human_auth_still_required(self): for headers in [{}, self.headers()]: for path in ['/api/miners/discover-one', '/api/tools/legacy-summary-refresh-one']: before = self.durable() self.denied(self.client.post(path, headers=headers, json={'ip_address': '198.51.100.111'}), before, 401) def test_csv_ip_only_requires_zone_even_with_one_global_match(self): self.miner(1, ip='198.51.100.112') before = self.durable() self.denied(self.upload('IP,Miner Type\n198.51.100.112,WRONG\n'), before, 422) def test_csv_same_ip_requires_namespace_and_exact_zone_updates_one(self): a, _ = self.miner(ip='198.51.100.126') b, _ = self.miner(1, ip='198.51.100.126') before = self.durable() self.denied(self.upload('IP,Miner Type\n198.51.100.126,WRONG\n'), before, 422) r = self.upload(f'IP,Agent Zone ID,Miner Type\n198.51.100.126,{self.zones[1]},VALID\n') self.assertEqual(r.status_code, 200, r.text) self.assertIsNone(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (a,))) self.assertEqual(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (b,)), 'VALID') self.assertEqual(before['miner_status_snapshots'], self.durable()['miner_status_snapshots']) def test_csv_duplicate_ip_inside_zone_rejected(self): self.miner(ip='198.51.100.130'); self.miner(ip='198.51.100.130') before = self.durable() self.denied(self.upload('IP,Miner Type\n198.51.100.130,WRONG\n', '&agent_zone_id=' + str(self.zones[0])), before, 409) def test_csv_exact_id_mac_serial_and_query_zone_legitimate(self): mid, mac = self.miner(ip='198.51.100.131') serial = uuid4().hex with connect() as c: c.execute('UPDATE miners SET miner_sn=%s WHERE id=%s', (serial, mid)) for heading, value in [('Miner ID', str(mid)), ('MAC', mac), ('Serial', serial), ('IP', '198.51.100.131')]: r = self.upload(f'{heading},Miner Type\n{value},VALID\n', '&agent_zone_id=' + str(self.zones[0])) self.assertEqual(r.status_code, 200, r.text) self.assertEqual(r.json()['data']['matched_count'], 1) def test_csv_conflicting_and_unknown_mac_never_fall_back_to_ip(self): a, _ = self.miner(ip='198.51.100.132') b, bm = self.miner(1, ip='198.51.100.133') for mac in [bm, '02:AA:BB:CC:DD:EF']: before = self.durable() self.denied(self.upload(f'Miner ID,IP,MAC,Miner Type\n{a},198.51.100.132,{mac},WRONG\n'), before, 409) before = self.durable() self.denied(self.upload(f'Miner ID,IP,Miner Type\n{a},198.51.100.133,WRONG\n'), before, 409) def test_csv_historical_mac_and_contradictory_namespace_serial_denied(self): mid, oldmac = self.miner(ip='198.51.100.134') with connect() as c: c.execute('UPDATE miners SET current_mac=%s,miner_sn=%s WHERE id=%s', ('02:' + ':'.join(f'{v:02X}' for v in uuid4().bytes[:5]), 'serial-' + str(mid), mid)) for columns, values in [ ('MAC', oldmac), ('Serial', 'unknown'), ('Site ID', str(self.sites[1])), ('Agent Zone ID', str(self.zones[1])), ('miner_id', str(uuid4())), ]: before = self.durable() self.denied(self.upload(f'Miner ID,{columns},Miner Type\n{mid},{values},WRONG\n'), before, 409) def test_csv_invalid_identifiers_and_query_row_disagreement(self): mid, _ = self.miner(ip='198.51.100.135') for header, value in [('MAC', 'bad-mac'), ('IP', '198.51.100.135/24'), ('Miner ID', 'bad-id'), ('Agent Zone ID', 'bad-zone')]: before = self.durable() self.denied(self.upload(f'{header},Miner Type\n{value},WRONG\n'), before, 422) before = self.durable() self.denied(self.upload(f'Miner ID,Agent Zone ID,Miner Type\n{mid},{self.zones[1]},WRONG\n', '&agent_zone_id=' + str(self.zones[0])), before, 409) def test_csv_unmatched_strong_identity_does_not_select_ip(self): self.miner(ip='198.51.100.136') before = self.durable() r = self.upload('MAC,IP,Miner Type\n02:AA:BB:CC:DD:EF,198.51.100.136,WRONG\n') self.assertEqual(r.status_code, 200, r.text) self.assertEqual(r.json()['data']['unmatched_count'], 1) self.assertEqual(before['miners'], self.durable()['miners']) def test_csv_batch_conflict_rolls_back_prior_valid_row(self): a, _ = self.miner(ip='198.51.100.137') b, _ = self.miner(1, ip='198.51.100.138') before = self.durable() self.denied(self.upload(f'Miner ID,IP,Miner Type\n{a},198.51.100.137,VALID\n{b},198.51.100.137,WRONG\n'), before, 409) def test_csv_preview_repeat_and_history_preservation(self): mid, _ = self.miner(ip='198.51.100.139') with connect() as c: c.execute('INSERT INTO miner_status_snapshots(id,miner_id,observed_at,mhs_av) VALUES (%s,%s,now(),123)', (uuid4(), mid)) before = self.durable() text = f'Miner ID,Miner Type\n{mid},VALID\n' r = self.client.post('/api/imports/whatsminer-tool-scan?dry_run=true&update_existing=true', headers=self.admin, files={'file': ('simulated.csv', text.encode(), 'text/csv')}) self.assertEqual(r.status_code, 200, r.text) self.assertEqual(before, self.durable()) for _ in range(2): self.assertEqual(self.upload(text).status_code, 200) self.assertEqual(before['miner_status_snapshots'], self.durable()['miner_status_snapshots']) self.assertEqual(before['miner_mac_ownerships'], self.durable()['miner_mac_ownerships']) def test_csv_actual_audit_permission_failure_rolls_back(self): mid, _ = self.miner(ip='198.51.100.140') role = 'b1_csv_' + uuid4().hex with connect() as c: c.execute('CREATE ROLE ' + role) c.execute('GRANT USAGE ON SCHEMA public TO ' + role) c.execute('GRANT SELECT,INSERT,UPDATE,DELETE ON ALL TABLES IN SCHEMA public TO ' + role) c.execute('REVOKE INSERT ON audit_logs FROM ' + role) def limited(): c = connect(); c.execute('SET ROLE ' + role); c.commit(); return c before = self.durable() try: with patch('app.main.get_db_connection', side_effect=limited): r = self.upload(f'Miner ID,Miner Type\n{mid},ROLLBACK\n') self.denied(r, before, 500) self.assertIn('audit_logs', r.text) finally: with connect() as c: c.execute('DROP OWNED BY ' + role); c.execute('DROP ROLE ' + role) def test_csv_concurrent_ownership_move_is_revalidated_after_real_lock_wait(self): mid, _ = self.miner(ip='198.51.100.141') owner = connect() owner.execute('UPDATE miners SET site_id=%s,agent_zone_id=%s,updated_at=clock_timestamp() WHERE id=%s', (self.sites[1], self.zones[1], mid)) with ThreadPoolExecutor(1) as pool: future = pool.submit(lambda: self.upload(f'Miner ID,Agent Zone ID,Miner Type\n{mid},{self.zones[0]},WRONG\n')) try: deadline = time.monotonic() + 4 waiting = 0 while time.monotonic() < deadline: waiting = self.scalar("SELECT count(*) FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock'") if waiting: break time.sleep(.02) self.assertGreater(waiting, 0) owner.commit() r = future.result(timeout=8) self.assertEqual(r.status_code, 409, r.text) self.assertEqual(self.scalar('SELECT agent_zone_id FROM miners WHERE id=%s', (mid,)), self.zones[1]) self.assertIsNone(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (mid,))) finally: owner.rollback(); owner.close() def test_csv_real_update_lock_prevents_mid_transaction_ownership_change(self): import app.discovery.csv_import as module mid, _ = self.miner(ip='198.51.100.142') entered, release = threading.Event(), threading.Event() original = module.resolve_inventory_row def gate(*args, **kwargs): row = original(*args, **kwargs); entered.set() self.assertTrue(release.wait(4)); return row def move(): with connect() as c: c.execute('UPDATE miners SET site_id=%s,agent_zone_id=%s,updated_at=clock_timestamp() WHERE id=%s', (self.sites[1], self.zones[1], mid)) with patch.object(module, 'resolve_inventory_row', side_effect=gate), ThreadPoolExecutor(2) as pool: upload = pool.submit(lambda: self.upload(f'Miner ID,Agent Zone ID,Miner Type\n{mid},{self.zones[0]},VALID\n')) self.assertTrue(entered.wait(4)); moving = pool.submit(move) try: deadline = time.monotonic() + 3 waiting = 0 while time.monotonic() < deadline: waiting = self.scalar("SELECT count(*) FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock'") if waiting: break time.sleep(.02) self.assertGreater(waiting, 0) finally: release.set() self.assertEqual(upload.result(timeout=8).status_code, 200) moving.result(timeout=8) self.assertEqual(self.scalar('SELECT agent_zone_id FROM miners WHERE id=%s', (mid,)), self.zones[1]) def test_csv_duplicate_headers_and_conflicting_aliases_rejected(self): mid, _ = self.miner(ip='198.51.100.143') for text, code in [ (f'Miner ID,Miner ID,Miner Type\n{mid},{mid},WRONG\n', 422), (f'Miner ID,miner_id,Miner Type\n{mid},{uuid4()},WRONG\n', 409), (f'Miner ID,Site ID,Agent Zone ID,Miner Type\n{mid},{self.sites[1]},{self.zones[0]},WRONG\n', 409), ]: before = self.durable() self.denied(self.upload(text), before, code) def test_csv_duplicate_serial_is_ambiguous(self): a, _ = self.miner(ip='198.51.100.144') b, _ = self.miner(1, ip='198.51.100.145') serial = 'duplicate-' + uuid4().hex with connect() as c: c.execute('UPDATE miners SET miner_sn=%s WHERE id IN (%s,%s)', (serial, a, b)) before = self.durable() self.denied(self.upload(f'Serial,Miner Type\n{serial},WRONG\n'), before, 409) def test_csv_global_mac_selection_with_overlapping_ip_is_valid(self): a, _ = self.miner(ip='198.51.100.146') b, mac = self.miner(1, ip='198.51.100.146') before = self.durable() r = self.upload(f'MAC,IP,Miner Type\n{mac},198.51.100.146,VALID\n') self.assertEqual(r.status_code, 200, r.text) self.assertIsNone(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (a,))) self.assertEqual(self.scalar('SELECT miner_type FROM miners WHERE id=%s', (b,)), 'VALID') after = self.durable() for table in ['miner_status_snapshots', 'miner_mac_ownerships', 'miner_network_identities']: self.assertEqual(before[table], after[table]) audit = self.scalar("SELECT new_value_json FROM audit_logs WHERE actor_user_id=%s AND action='import.whatsminertool_csv.applied' ORDER BY created_at DESC LIMIT 1", (self.uid,)) self.assertEqual(audit['updates'][0]['miner_id'], str(b)) self.assertEqual(audit['updates'][0]['agent_zone_id'], str(self.zones[1])) def test_csv_read_only_user_cannot_apply_metadata(self): uid = uuid4() with connect() as c: c.execute("INSERT INTO users(id,username,display_name,password_hash,role_id) SELECT %s,%s,'Read only','unused',id FROM roles WHERE name='Read Only'", (uid, str(uid))) headers = {'Authorization': 'Bearer ' + create_access_token(str(uid), str(uid), 'Read Only')} mid, _ = self.miner(ip='198.51.100.147') before = self.durable() self.denied(self.upload(f'Miner ID,Miner Type\n{mid},WRONG\n', headers=headers), before, 403)