#!/usr/bin/env python3 ########################################################################### ## Copyright (C) Wizardry and Steamworks 2026 - License: MIT ## ########################################################################### # --- INTEGRITY STAMP --- try: import hashlib from pathlib import Path print(f"\n[INTEGRITY] RUNNING SCRIPT SHA-256 DIGEST: {hashlib.sha256(Path(__file__).read_bytes()).hexdigest().upper()}\n", flush=True) except Exception: pass import os import sys import time import signal import subprocess import asyncio import json import base64 import traceback import shutil import random import uuid import atexit import re import socket import gc from pathlib import Path from typing import Optional, Dict, Any, List, Tuple from dataclasses import dataclass, field from enum import Enum from datetime import datetime, timezone, timedelta import structlog import httpx import websockets from stem.control import Controller import psutil # Fix for Python 3.10 and older - define UTC timezone try: UTC = timezone.UTC except AttributeError: UTC = timezone(timedelta(0)) structlog.configure( processors=[ structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer() ] ) logger = structlog.get_logger() # Sanitization ANSI_ESCAPE = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') CONTROL_CHARS = re.compile(r'[\x00-\x1f\x7f]') MULTI_SPACE = re.compile(r'\s+') TAB_CHARS = re.compile(r'\t+') MARTIAN_CHARS = re.compile(r'[^\x20-\x7E]') def sanitize_log_line(text): if not text: return "" text = ANSI_ESCAPE.sub('', text) text = CONTROL_CHARS.sub('', text) text = TAB_CHARS.sub(' ', text) text = MARTIAN_CHARS.sub('', text) text = MULTI_SPACE.sub(' ', text) return text.strip() def is_monerod_spam(line): if not line: return True spam_patterns = [ 'Including transaction', 'Received new tx while syncing', 'LEVIN_PACKET', 'bytes received for category', 'NOTIFY_NEW_TRANSACTIONS', 'partial msg received', 'Partial result', 'DROP.*CONNECTION', 'CLOSE CONNECTION', 'Destructing connection', 'JSON RPC request', 'COMMAND_TIMED_SYNC', 'REMOTE PEERLIST', 'getblocks.bin', 'get_info' ] for pattern in spam_patterns: if re.search(pattern, line, re.IGNORECASE): return True return False def hlog(comp, msg): if not msg: return clean_msg = sanitize_log_line(msg) if not clean_msg: return if comp == "monerod" and is_monerod_spam(clean_msg): return logger.info("legacy_log", component=comp, message=clean_msg) def print_raw(msg): print(msg, flush=True) def encode_base64(data): if data is None: return None if isinstance(data, bytes): return base64.b64encode(data).decode('ascii') return base64.b64encode(data.encode()).decode('ascii') def decode_base64(data): if data is None: return None return base64.b64decode(data) # ============================================================================ # I2P BASE64 HANDLING - URL-safe variant with ~ and - # ============================================================================ def decode_i2p_base64(data): """Decode I2P Base64 (URL-safe variant with ~ and -).""" if data is None: return None if isinstance(data, bytes): return data if not isinstance(data, str): data = str(data) data = data.strip() if not data: return None # Try 1: URL-safe base64 (~ -> /, - -> +) try: converted = data.replace('~', '/').replace('-', '+') converted = re.sub(r'[^A-Za-z0-9+/=]', '', converted) while len(converted) % 4 != 0: converted += '=' return base64.b64decode(converted) except Exception: pass # Try 2: Standard base64 with padding try: cleaned = re.sub(r'[^A-Za-z0-9+/=]', '', data) while len(cleaned) % 4 != 0: cleaned += '=' return base64.b64decode(cleaned) except Exception: pass # Try 3: Extract valid base64 substring try: match = re.search(r'[A-Za-z0-9+/]{20,}={0,2}', data.replace('~', '/').replace('-', '+')) if match: valid_part = match.group(0) while len(valid_part) % 4 != 0: valid_part += '=' return base64.b64decode(valid_part) except Exception: pass hlog("I2P_BASE64_ERR", f"Failed to decode I2P Base64: {data[:50]}... (will generate via SAM)") return None def encode_i2p_base64(data): """Encode I2P key to URL-safe base64 (with ~ and -).""" if data is None: return None if isinstance(data, str): data = data.encode() b64 = base64.b64encode(data).decode('ascii') return b64.replace('+', '-').replace('/', '~') # ============================================================================ # CONFIGURATION - Static ports # ============================================================================ TOR_SOCKS_PORT = 9050 TOR_CONTROL_PORT = 9051 I2P_SAM_PORT = 7656 I2P_HTTP_PORT = 7657 MONEROD_P2P_PORT = 28080 MONEROD_P2P_INTERNAL_PORT = 28084 MONEROD_RPC_MAIN_PORT = 18081 MONEROD_RPC_RESTRICTED_PORT = 18082 WALLET_RPC_PORT = 28085 WS_INTERNAL_PORT = 8080 P2P_EXT_PORT = 28080 RPC_BIND_IP = "127.0.0.1" P2P_BIND_IP = "127.0.0.1" LOOPBACK_IP = "127.0.0.1" BIN_GPG = "/usr/bin/gpg" BIN_CURL = "/usr/bin/curl" BIN_CRYPTSETUP = "/sbin/cryptsetup" BIN_DMSETUP = "/sbin/dmsetup" BIN_MOUNT = "/bin/mount" BIN_UMOUNT = "/bin/umount" BIN_TOR = "/usr/bin/tor" BIN_JAVA = "/usr/bin/java" BIN_I2PD = "/usr/sbin/i2pd" BIN_MONEROD = "/usr/bin/monerod" BIN_XMR_RPC = "/usr/bin/monero-wallet-rpc" BIN_IPTABLES = "/usr/sbin/iptables" BIN_CHOWN = "/usr/bin/chown" BIN_CHMOD = "/usr/bin/chmod" BIN_SUDO = "/usr/bin/sudo" BIN_XMRIG_PROXY = "/usr/local/bin/xmrig-proxy" I2P_HOME = "/opt/i2p" I2P_DATA_DIR = "/var/lib/i2p" I2P_CONFIG_DIR = "/var/lib/i2p/i2p-config" LUKS_DEVICE = "/dev/xmr-raw" LUKS_NAME = "monero_crypt" LUKS_TYPE = "plain" LUKS_CIPHER = "aes-xts-plain64" LUKS_KEY_SIZE = "256" LUKS_HASH = "plain" BOOTSTRAP_ENCRYPTION_KEY = "EFA3B2C5B8DEA6BF824C82543DE933083623DDF1" ONION_PEERS = [ "moneroxm6atvba6d75vpx6nqy4asvpgre6kwj5o2v6li7st3kj6747ad.onion:18080", "sq67vyo7vskz5ckm5v6id7zuv6x6pxtxkyzctkizvxy2uio5u25oymad.onion:18083", "vww6ybal4bd7szmgncyruucpgfkqahzddi37ktbh3cs6v76twv7xq7ad.onion:18080", "nodesv3uzvj7rvyly675mzo7hsc6kxgvsl26z64nkpdq73wzly4mnyyd.onion:18081", "monero7zscbby7v6asx36o7625k44634t45vqyhk47p7lsq7f74pnyid.onion:18080", "cakexmrl7bonq7ovjka5kuwuyd3f7qnkz6z6s6dmsy3uckwra7bvggyd.onion:18081", "fz2lbxvjob6ifeonngaep2xvf2ypxjjn23i3ncblcxjreovev56ubyyd.onion:18089", "a6orjo6aiotog3njppja5jwnd3rexzfjiejxnojvw74p3kma45fundid.onion:18089", "f5brpwb6y6sjkuxq2jt6avpufjdayqpiy3gkbmeuclw4olt7wodfexid.onion:18089", "csxmritzk2qdgqmou2vwyrwu65xabimvmeniestaartks4fhlocfoeyd.onion:18081", "sfprpc5klzs5vyitq2mrooicgk2wcs5ho2nm3niqduvzn5o6ylaslaqd.onion:18089", "rucknium757bokwv3ss35ftgc3gzb7hgbvvglbg3hisp7tsj2fkd2nyd.onion:18081", "plowsof3t5hogddwabaeiyrno25efmzfxyro2vligremt7sxpsclfaid.onion:18089", "plowsoffjexmxalw73tkjmf422gq6575fc7vicuu4javzn2ynnte6tyd.onion:18089", "plowsofe6cleftfmk2raiw5h2x66atrik3nja4bfd3zrfa2hdlgworad.onion:18089", "i4jsfwmw22yjzzmzkoc7aahiaqlyhnykn5wxel43u3o5ibz2k4275jqd.onion:18081", "6dsdenp6vjkvqzy4wzsnzn6wixkdzihx3khiumyzieauxuxslmcaeiad.onion:18081", ] DATA_DIR = "/mnt/monero-secure" HS_DIR = "/dev/shm/tor/hs" I2P_PERSISTENT_DIR = "/dev/shm/i2pd" RINGDB_DIR = "/dev/shm/shared-ringdb" WALLET_CACHE_DIR = os.path.join(DATA_DIR, "wallet_cache") SYNC_MAX_ATTEMPTS = 120 BLOCK_TEMPLATE_MAX_RETRIES = 30 BLOCK_TEMPLATE_RETRY_DELAY = 10 RPC_READY_MAX_ATTEMPTS = 60 TOR_MAX_WAIT = 120 I2P_MAX_WAIT = 900 DIFFICULTY_INITIAL = 100000 DIFFICULTY_ADJUST_INTERVAL = 300 NONCE_RANGE_MULTIPLIER_BASE = 10000000 hlog("PORT_CONFIG", f"P2P port: {MONEROD_P2P_PORT}") hlog("PORT_CONFIG", f"RPC ports: {MONEROD_RPC_MAIN_PORT}/{MONEROD_RPC_RESTRICTED_PORT}") hlog("PORT_CONFIG", f"Wallet RPC port: {WALLET_RPC_PORT}") hlog("PORT_CONFIG", f"I2P HTTP console: {I2P_HTTP_PORT}") hlog("PORT_CONFIG", f"I2P data dir: {I2P_PERSISTENT_DIR}") # ============================================================================ # STATE MANAGEMENT # ============================================================================ class ServiceStatus(Enum): STOPPED = "stopped" STARTING = "starting" RUNNING = "running" DEGRADED = "degraded" FAILED = "failed" @dataclass class ServiceState: status: ServiceStatus = ServiceStatus.STOPPED last_error: Optional[str] = None last_check: float = 0 ready: bool = False STATE = { "luks_key": None, "recovery": {"seed": None, "restore_height": 0}, "tor": {"secret_key_b64": None, "onion": None}, "i2p": { "destination_b64": None, "router_keys_b64": None, "reseed_su3_b64": None, "reseed_timestamp": 0, "destination": None, "tunnels_ready": False, "peer_count": 0, "tunnel_count": 0 }, "primary_address": None, "derived_addresses": [] } RUNNING = True PROCESSES: Dict[str, asyncio.subprocess.Process] = {} TASKS: List[asyncio.Task] = [] CLIENT_SESSIONS = {} RECYCLED_NONCE_RANGES = [] NEXT_RANGE_MULTIPLIER = 1 MINING_JOBS = {} CURRENT_JOB_ID = None DIFFICULTY = DIFFICULTY_INITIAL TOTAL_HASHRATE = 0 LAST_DIFFICULTY_ADJUST = time.time() MONEROD_SYNCED = False MONEROD_RPC_READY = False WALLET_RPC_STARTED = False RESTART_COUNTERS = {"wallet_rpc": 0, "monerod": 0, "tor": 0, "i2p": 0} LAST_RESTART_TIME = {"wallet_rpc": 0, "monerod": 0, "tor": 0, "i2p": 0} TEMP_WALLET_DIR = None I2P_DESTINATION = None I2P_PRIVATE_KEY = None TOR_ONION = None TOR_PRIVATE_KEY = None I2P_ROUTER_KEYS = None LAST_RESEED_GENERATE = 0 TOR_READY = False I2P_READY = False ACTIVE_CLIENTS = 0 I2P_TUNNEL_COUNT = 0 I2P_PEER_COUNT = 0 I2P_NETWORK_OK = False STAGE_COMPLETE = set() WALLET_PASSWORD = None RESTORE_HEIGHT = 0 def derive_wallet_password(seed: str) -> str: """Derive a wallet password from the seed. Only in memory.""" return hashlib.sha256(seed.encode()).hexdigest()[:32] # ============================================================================ # PGP STATE BUNDLING - COMPLETE BOOTSTRAP JSON # ============================================================================ async def emit_encrypted_state(label, state_snapshot=None): global STAGE_COMPLETE, RESTORE_HEIGHT, TOR_ONION, TOR_PRIVATE_KEY, I2P_DESTINATION, I2P_PRIVATE_KEY if label in STAGE_COMPLETE: return True try: if state_snapshot is None: state_snapshot = STATE.copy() # Ensure restore_height is preserved if "recovery" in state_snapshot and state_snapshot["recovery"].get("restore_height", 0) == 0: if RESTORE_HEIGHT > 0: state_snapshot["recovery"]["restore_height"] = RESTORE_HEIGHT # Build COMPLETE bootstrap.json structure complete_bootstrap = { "luks_key": state_snapshot.get("luks_key"), "tor": { "secret_key_b64": state_snapshot.get("tor", {}).get("secret_key_b64") or TOR_PRIVATE_KEY, "onion": state_snapshot.get("tor", {}).get("onion") or TOR_ONION }, "i2p": { "destination_b64": state_snapshot.get("i2p", {}).get("destination_b64"), "router_keys_b64": state_snapshot.get("i2p", {}).get("router_keys_b64"), "reseed_su3_b64": state_snapshot.get("i2p", {}).get("reseed_su3_b64"), "destination": state_snapshot.get("i2p", {}).get("destination") or I2P_DESTINATION }, "recovery": { "seed": state_snapshot.get("recovery", {}).get("seed"), "restore_height": state_snapshot.get("recovery", {}).get("restore_height") or RESTORE_HEIGHT }, "primary_address": state_snapshot.get("primary_address"), "derived_addresses": state_snapshot.get("derived_addresses", []) } # Clean bundle - just stage, timestamp, and bootstrap bundle = { "stage": label, "timestamp": datetime.now(UTC).isoformat(), "bootstrap": complete_bootstrap } gpg_cmd = [ BIN_GPG, '--batch', '--quiet', '--encrypt', '--recipient', BOOTSTRAP_ENCRYPTION_KEY, '--armor', '--always-trust' ] proc = await asyncio.create_subprocess_exec( *gpg_cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await proc.communicate(input=json.dumps(bundle, indent=2).encode()) if proc.returncode == 0: print_raw(f"\n[{label}] --- STATE BUNDLE START ---\n{stdout.decode().strip()}\n[{label}] --- STATE BUNDLE END ---\n") hlog("STATE", f"Emitted encrypted state bundle: {label}") STAGE_COMPLETE.add(label) return True else: error_msg = f"GPG encryption failed for {label}: {stderr.decode()}" hlog("STATE_CRITICAL", error_msg) await shutdown() sys.exit(1) except Exception as e: error_msg = f"Failed to emit state {label}: {e}" hlog("STATE_CRITICAL", error_msg) await shutdown() sys.exit(1) # ============================================================================ # EMIT FINAL STATE - Complete bootstrap.json at shutdown # ============================================================================ async def emit_final_state(): """Emit the complete final state as a full bootstrap.json payload.""" global TOR_ONION, TOR_PRIVATE_KEY, I2P_DESTINATION, I2P_PRIVATE_KEY, RESTORE_HEIGHT # Build complete bootstrap.json from current state complete_bootstrap = { "luks_key": STATE.get("luks_key"), "tor": { "secret_key_b64": TOR_PRIVATE_KEY or STATE.get("tor", {}).get("secret_key_b64"), "onion": TOR_ONION or STATE.get("tor", {}).get("onion") }, "i2p": { "destination_b64": STATE.get("i2p", {}).get("destination_b64") or encode_i2p_base64(I2P_PRIVATE_KEY) if I2P_PRIVATE_KEY else None, "router_keys_b64": STATE.get("i2p", {}).get("router_keys_b64"), "reseed_su3_b64": STATE.get("i2p", {}).get("reseed_su3_b64"), "destination": I2P_DESTINATION or STATE.get("i2p", {}).get("destination") }, "recovery": { "seed": STATE.get("recovery", {}).get("seed"), "restore_height": RESTORE_HEIGHT or STATE.get("recovery", {}).get("restore_height", 0) }, "primary_address": STATE.get("primary_address"), "derived_addresses": STATE.get("derived_addresses", []) } # Remove None values to match bootstrap.json format complete_bootstrap = {k: v for k, v in complete_bootstrap.items() if v is not None} if "tor" in complete_bootstrap: complete_bootstrap["tor"] = {k: v for k, v in complete_bootstrap["tor"].items() if v is not None} if "i2p" in complete_bootstrap: complete_bootstrap["i2p"] = {k: v for k, v in complete_bootstrap["i2p"].items() if v is not None} if "recovery" in complete_bootstrap: complete_bootstrap["recovery"] = {k: v for k, v in complete_bootstrap["recovery"].items() if v is not None} bundle = { "stage": "FINAL_STATE", "timestamp": datetime.now(UTC).isoformat(), "bootstrap": complete_bootstrap } try: gpg_cmd = [ BIN_GPG, '--batch', '--quiet', '--encrypt', '--recipient', BOOTSTRAP_ENCRYPTION_KEY, '--armor', '--always-trust' ] proc = await asyncio.create_subprocess_exec( *gpg_cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await proc.communicate(input=json.dumps(bundle, indent=2).encode()) if proc.returncode == 0: print_raw(f"\n[FINAL_STATE] --- STATE BUNDLE START ---\n{stdout.decode().strip()}\n[FINAL_STATE] --- STATE BUNDLE END ---\n") hlog("STATE", "Emitted final complete bootstrap.json bundle") return True else: hlog("STATE_ERR", f"Final state encryption failed: {stderr.decode()}") return False except Exception as e: hlog("STATE_ERR", f"Failed to emit final state: {e}") return False # ============================================================================ # CBF (Compressed Brainfuck) Implementation # ============================================================================ def compress_to_cbf(bf_code): if not bf_code: return "" compressed = [] i = 0 n = len(bf_code) while i < n: char = bf_code[i] if char in '+-<>.,': j = i while j < n and bf_code[j] == char: j += 1 count = j - i if count > 1: compressed.append(str(count) + char) else: compressed.append(char) i = j elif char == '[': depth = 1 j = i + 1 while j < n and depth > 0: if bf_code[j] == '[': depth += 1 elif bf_code[j] == ']': depth -= 1 j += 1 inner = compress_to_cbf(bf_code[i+1:j-1]) compressed.append('[' + inner + ']') i = j else: compressed.append(char) i += 1 return ''.join(compressed) def decompress_cbf(cbf_code): result = [] i = 0 n = len(cbf_code) while i < n: if cbf_code[i].isdigit(): num_start = i while i < n and cbf_code[i].isdigit(): i += 1 N = int(cbf_code[num_start:i]) if i < n: token = cbf_code[i] if token in '+-<>.,': result.append(token * N) i += 1 elif token == '[': depth = 1 j = i + 1 while j < n and depth > 0: if cbf_code[j] == '[': depth += 1 elif cbf_code[j] == ']': depth -= 1 j += 1 body = cbf_code[i+1:j-1] decompressed_body = decompress_cbf(body) result.append(decompressed_body * N) i = j else: i += 1 else: if cbf_code[i] == '[': depth = 1 j = i + 1 while j < n and depth > 0: if cbf_code[j] == '[': depth += 1 elif cbf_code[j] == ']': depth -= 1 j += 1 inner = cbf_code[i+1:j-1] result.append('[' + decompress_cbf(inner) + ']') i = j else: result.append(cbf_code[i]) i += 1 return ''.join(result) def generate_bf_log(data): diagnostic = { "type": "AI_DIAGNOSTIC", "format": "cBF_compressed_brainfuck", "parser": "Use decompress_cbf() then run BF interpreter", "instructions": "cBF grammar: I ::= ... | N Token | N '[' I* ']'", "data": data } payload = json.dumps(diagnostic, separators=(',', ':')) bf, cur = "", 0 for char in payload: target = ord(char) diff = target - cur bf += ("+" if diff > 0 else "-") * abs(diff) + "." cur = target cbf = compress_to_cbf(bf) return f"\\n--- AI DIAGNOSTIC BEGIN (cBF compressed) ---\\n{cbf}\\n--- AI DIAGNOSTIC END ---\\n" # ============================================================================ # HELPER FUNCTIONS # ============================================================================ def is_port_available(host, port): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((host, port)) sock.close() return True except OSError: return False except Exception: return True # ============================================================================ # RINGDB SETUP - FIXED PERMISSIONS # ============================================================================ def setup_ringdb(): """Create and set up shared ringdb directory with proper permissions.""" try: os.makedirs(RINGDB_DIR, mode=0o777, exist_ok=True) # Fix permissions - make it world-writable since we're running as root try: subprocess.run(["chmod", "-R", "777", RINGDB_DIR], check=False) # Also chown to root since we're running as root subprocess.run(["chown", "-R", "root:root", RINGDB_DIR], check=False) except Exception as e: hlog("RINGDB_WARN", f"Could not set permissions: {e}") hlog("RINGDB", f"RingDB directory ready: {RINGDB_DIR}") return True except Exception as e: hlog("RINGDB_ERR", f"Failed to setup ringdb: {e}") return False def cleanup_ringdb(): """Clean up RingDB directory and remove stale locks.""" try: if os.path.exists(RINGDB_DIR): # Remove lock files for f in os.listdir(RINGDB_DIR): if f.endswith('.lock') or f == 'lock.mdb': try: os.remove(os.path.join(RINGDB_DIR, f)) except: pass # Remove the directory contents but keep the directory shutil.rmtree(RINGDB_DIR, ignore_errors=True) os.makedirs(RINGDB_DIR, mode=0o777, exist_ok=True) subprocess.run(["chmod", "-R", "777", RINGDB_DIR], check=False) subprocess.run(["chown", "-R", "root:root", RINGDB_DIR], check=False) except Exception: pass # ============================================================================ # MONEROD SYNC FUNCTION # ============================================================================ async def is_monerod_synced(): """Check if monerod is fully synced and RPC is responsive.""" try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.post( f"http://{RPC_BIND_IP}:{MONEROD_RPC_MAIN_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_info"}, timeout=10.0 ) if resp.status_code == 200: data = resp.json().get("result", {}) # Check both sync status and that we have valid data is_synced = data.get("synchronized", False) height = data.get("height", 0) target_height = data.get("target_height", 0) # If target_height is 0, assume we're synced (daemon just started) if target_height == 0 and height > 0: return True return is_synced or (target_height > 0 and height >= target_height - 5) except (httpx.TimeoutException, httpx.ConnectError) as e: hlog("MONEROD_SYNC", f"RPC not ready yet: {e}") except Exception as e: hlog("MONEROD_SYNC", f"Sync check error: {e}") return False # ============================================================================ # I2P STATISTICS COLLECTION # ============================================================================ async def get_i2p_console_stats(): try: async with httpx.AsyncClient( timeout=5.0, follow_redirects=True, headers={ 'User-Agent': 'Mozilla/5.0 (compatible; I2P-Monitor/1.0)', 'Accept': 'text/html,application/xhtml+xml', 'Connection': 'close' } ) as client: urls = [ f"http://127.0.0.1:{I2P_HTTP_PORT}/console", f"http://127.0.0.1:{I2P_HTTP_PORT}/home", f"http://127.0.0.1:{I2P_HTTP_PORT}/" ] html = None for url in urls: try: resp = await client.get(url) if resp.status_code == 200: html = resp.text break except Exception: continue if not html: return {"status": "no_content"} stats = { "peers": 0, "tunnels": 0, "status": "unknown", "uptime": "unknown", "version": "unknown", "active_peers": 0, "known_peers": 0, "tunnel_count": 0 } if "Network: OK" in html: stats["status"] = "running" elif "Network: Firewalled" in html: stats["status"] = "firewalled" elif "Rejecting tunnels" in html: stats["status"] = "starting" elif "Testing" in html or "Connecting" in html: stats["status"] = "connecting" elif "Reseeding" in html: stats["status"] = "reseeding" active_match = re.search(r'Active:]*>(\d+)\s*/\s*(\d+)', html, re.IGNORECASE) if active_match: stats["active_peers"] = int(active_match.group(1)) stats["peers"] = int(active_match.group(1)) known_match = re.search(r'Known:]*>(\d+)', html, re.IGNORECASE) if known_match: stats["known_peers"] = int(known_match.group(1)) tunnel_match = re.search(r'Client:]*>(\d+)', html, re.IGNORECASE) if tunnel_match: stats["tunnels"] = int(tunnel_match.group(1)) else: tunnel_match = re.search(r'Tunnels?]*>(\d+)', html, re.IGNORECASE) if tunnel_match: stats["tunnels"] = int(tunnel_match.group(1)) uptime_match = re.search(r'Uptime:]*>(\d+)\s*min', html, re.IGNORECASE) if uptime_match: stats["uptime"] = f"{uptime_match.group(1)} min" else: uptime_match = re.search(r'Uptime:]*>([\d:]+)', html, re.IGNORECASE) if uptime_match: stats["uptime"] = uptime_match.group(1) version_match = re.search(r'Version:]*>([\d.]+)', html, re.IGNORECASE) if version_match: stats["version"] = version_match.group(1) if "shared clients" in html and "Ready" in html: stats["status"] = "running" return stats except Exception as e: hlog("I2P_CONSOLE_ERR", f"Console fetch error: {str(e)[:100]}") return None return None # ============================================================================ # I2P LOG STREAMING # ============================================================================ async def stream_i2p_logs(stream): try: while True: line = await stream.readline() if not line: break clean_line = sanitize_log_line(line.decode().strip()) if "error" in clean_line.lower() or "warn" in clean_line.lower(): hlog("i2p", clean_line) continue important_patterns = [ "router started", "reseeding", "starting up", "tunnel", "peer", "connected", "network ok", "building", "established", "destination", "listening", "ready", "failed", "error", "netdb", "ntcp", "ssu", "transports", "clients", "addressbook", "console", "i2np", "bandwidth", "floodfill" ] show = False for pattern in important_patterns: if pattern in clean_line.lower(): show = True break if "I2P is starting" in clean_line or "Starting I2P" in clean_line: hlog("I2P", "I2P Router is starting up...") continue elif "Reseeding" in clean_line: hlog("I2P", "Reseeding from network...") continue elif "Network ready" in clean_line or "Router is ready" in clean_line: hlog("I2P", "I2P Network is ready!") continue if show: hlog("i2p", clean_line) except asyncio.CancelledError: pass # ============================================================================ # I2P FUNCTIONS - SAM bridge only # ============================================================================ async def setup_i2p_directories(): """Create all necessary I2P directories with proper permissions.""" dirs = [ I2P_PERSISTENT_DIR, f"{I2P_PERSISTENT_DIR}/netDb", f"{I2P_PERSISTENT_DIR}/addressbook", f"{I2P_PERSISTENT_DIR}/certificates", f"{I2P_PERSISTENT_DIR}/destinations/private", f"{I2P_PERSISTENT_DIR}/private", f"{I2P_PERSISTENT_DIR}/run", I2P_CONFIG_DIR, f"{I2P_CONFIG_DIR}/destinations/private", f"{I2P_CONFIG_DIR}/private", f"{I2P_CONFIG_DIR}/clients.config.d", I2P_DATA_DIR, ] for d in dirs: try: os.makedirs(d, mode=0o755, exist_ok=True) except Exception as e: hlog("I2P_WARN", f"Could not create directory structural node {d}: {e}") try: for root_dir in set([I2P_PERSISTENT_DIR, I2P_CONFIG_DIR, I2P_DATA_DIR]): if os.path.exists(root_dir): subprocess.run(["chown", "-R", "i2puser:i2puser", root_dir], check=True, capture_output=True) subprocess.run(["chmod", "-R", "755", root_dir], check=True, capture_output=True) hlog("I2P", "I2P tree structured and fully assigned to i2puser:i2puser") except subprocess.CalledProcessError as e: hlog("I2P_WARN", f"Permissions enforcement sub-pass encountered an error: {e.stderr.decode().strip()}") except Exception as e: hlog("I2P_WARN", f"Unexpected error during permission synchronization loop: {e}") async def create_i2p_config(): """Create I2P configuration files with SAM bridge.""" config_dir = I2P_CONFIG_DIR os.makedirs(config_dir, exist_ok=True) wrapper_config = f"""# I2P Wrapper Configuration - Generated by orchestrator i2p.config.dir={I2P_CONFIG_DIR} i2p.data.dir={I2P_DATA_DIR} i2p.router.data.dir={I2P_DATA_DIR} i2p.home.dir={I2P_HOME} """ wrapper_path = os.path.join(config_dir, "wrapper.config") try: with open(wrapper_path, 'w') as f: f.write(wrapper_config) hlog("I2P", f"Created wrapper config: {wrapper_path}") except Exception as e: hlog("I2P_WARN", f"Could not write wrapper config: {e}") router_config = f"""# I2P Router Configuration - Generated by orchestrator i2p.router.webconsole.port={I2P_HTTP_PORT} i2p.router.webconsole.address=127.0.0.1 i2p.router.network.interface=127.0.0.1 i2p.router.transport.ntcp.ip=127.0.0.1 i2p.router.transport.ssu.ip=127.0.0.1i2p.router.data.dir={I2P_DATA_DIR} i2p.router.data.dir2={I2P_DATA_DIR} i2p.router.identity=default i2p.router.identity.public=default """ router_path = os.path.join(config_dir, "router.config") try: with open(router_path, 'w') as f: f.write(router_config) hlog("I2P", f"Created router config: {router_path}") except Exception as e: hlog("I2P_WARN", f"Could not write router config: {e}") clients_path = os.path.join(config_dir, "clients.config") if not os.path.exists(clients_path): hlog("I2P_WARN", f"clients.config missing from runtime directory! Regenerating lifecycle threads...") clients_config = f"""# I2P Clients Configuration - Re-generated by orchestrator clientApp.0.main=net.i2p.router.web.RouterConsoleRunner clientApp.0.name=RouterConsole clientApp.0.args=-p {I2P_HTTP_PORT} -h 127.0.0.1 ./webapps/ clientApp.0.delay=5 clientApp.0.startOnLoad=true clientApp.1.main=net.i2p.sam.SAMBridge clientApp.1.name=SAMBridge clientApp.1.args=sam.keys 127.0.0.1 {I2P_SAM_PORT} clientApp.1.delay=10 clientApp.1.startOnLoad=true """ try: with open(clients_path, 'w') as f: f.write(clients_config) hlog("I2P", f"Successfully restored clients.config: {clients_path}") except Exception as e: hlog("I2P_WARN", f"Could not enforce clients configuration persistence: {e}") try: for cfg_file in [wrapper_path, router_path, clients_path]: if os.path.exists(cfg_file): shutil.chown(cfg_file, user="i2puser", group="i2puser") os.chmod(cfg_file, 0o644) except Exception as e: hlog("I2P_WARN", f"Failed running auxiliary permission pass on config artifacts: {e}") async def get_i2p_destination_via_sam(): """Get I2P destination via SAM bridge (auto-starts SAM if needed).""" import socket import re import base64 hlog("I2P_DEBUG", "Attempting to get I2P destination via SAM bridge...") try: # Check if SAM is running sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2) result = sock.connect_ex(('127.0.0.1', 7656)) sock.close() if result != 0: hlog("I2P", "SAM bridge not running, starting it...") cmd = [ BIN_SUDO, "-u", "i2puser", BIN_JAVA, "-cp", f"{I2P_HOME}/lib/*", f"-Di2p.config.dir={I2P_CONFIG_DIR}", f"-Di2p.data.dir={I2P_DATA_DIR}", "net.i2p.sam.SAMBridge", "sam.keys", "127.0.0.1", "7656" ] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT ) PROCESSES["sam"] = proc TASKS.append(asyncio.create_task(stream_logs("sam", proc.stdout))) hlog("I2P", f"SAM bridge started with PID: {proc.pid}") await asyncio.sleep(5) else: hlog("I2P_DEBUG", "SAM bridge already running") # Connect to SAM and get destination s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(5) s.connect(('127.0.0.1', 7656)) s.send(b"HELLO VERSION MIN=3.1 MAX=3.1\n") response = s.recv(1024) if b"OK" not in response: hlog("I2P_SAM_ERR", f"SAM HELLO failed: {response}") s.close() return None s.send(b"DEST GENERATE\n") response = s.recv(1024) s.close() response_str = response.decode() pub_match = re.search(r'PUB=([A-Za-z0-9~_-]+)', response_str) if pub_match: destination = pub_match.group(1) hlog("I2P_DEBUG", f"Got I2P destination via SAM: {destination[:30]}...") STATE["i2p"]["destination"] = destination STATE["i2p"]["destination_b64"] = destination dest_bytes = decode_i2p_base64(destination) if dest_bytes is None: hlog("I2P_SAM_ERR", f"Failed to decode I2P destination: {destination[:30]}...") return destination for path in [ f"{I2P_CONFIG_DIR}/destinations/monerod.dat", f"{I2P_CONFIG_DIR}/private/monerod.dat", f"{I2P_DATA_DIR}/destinations/monerod.dat", f"{I2P_DATA_DIR}/private/monerod.dat", f"{I2P_PERSISTENT_DIR}/destinations/monerod.dat", f"{I2P_PERSISTENT_DIR}/private/monerod.dat", ]: try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, 'wb') as f: f.write(dest_bytes) hlog("I2P_DEBUG", f"Saved destination to {path}") except Exception as e: hlog("I2P_DEBUG", f"Could not save to {path}: {e}") await emit_encrypted_state("I2P_KEY_CAPTURED_VIA_SAM") return destination except Exception as e: hlog("I2P_SAM_ERR", f"SAM bridge error: {e}") return None return None async def get_i2p_destination(): """Get I2P destination and private key from files.""" global I2P_PRIVATE_KEY, I2P_DESTINATION hlog("I2P_DEBUG", "Attempting to get I2P destination from files...") key_locations = [ f"{I2P_CONFIG_DIR}/destinations/monerod.dat", f"{I2P_CONFIG_DIR}/private/monerod.dat", f"{I2P_CONFIG_DIR}/destinations/private/monerod.dat", f"{I2P_PERSISTENT_DIR}/destinations/monerod.dat", f"{I2P_PERSISTENT_DIR}/private/monerod.dat", f"{I2P_PERSISTENT_DIR}/destinations/private/monerod.dat", f"{I2P_DATA_DIR}/destinations/monerod.dat", f"{I2P_DATA_DIR}/private/monerod.dat", ] for key_path in key_locations: if os.path.exists(key_path): hlog("I2P_DEBUG", f"Checking key file: {key_path}") try: with open(key_path, 'rb') as f: key_data = f.read() if key_data and len(key_data) > 30: STATE["i2p"]["destination_b64"] = encode_i2p_base64(key_data) I2P_PRIVATE_KEY = key_data hlog("I2P_DEBUG", f"Found I2P private key at {key_path} ({len(key_data)} bytes)") try: with open(key_path, 'r') as f_text: content = f_text.read().strip() if content and len(content) > 50 and not content.startswith('-----'): I2P_DESTINATION = content STATE["i2p"]["destination"] = content hlog("I2P_DEBUG", f"Found I2P destination as text: {content[:30]}...") except UnicodeDecodeError: pass await emit_encrypted_state("I2P_KEY_CAPTURED") return I2P_DESTINATION except Exception as e: hlog("I2P_DEBUG", f"Could not read {key_path}: {e}") if I2P_PRIVATE_KEY and not I2P_DESTINATION: I2P_DESTINATION = encode_i2p_base64(I2P_PRIVATE_KEY) STATE["i2p"]["destination"] = I2P_DESTINATION hlog("I2P_DEBUG", "Derived I2P destination from private key") return I2P_DESTINATION hlog("I2P_WARN", "No I2P private key found in any location!") return None async def wait_for_i2p_circuit(): """Wait for I2P to be ready and capture the valid destination key.""" global I2P_READY, I2P_DESTINATION, I2P_NETWORK_OK, I2P_PEER_COUNT, I2P_TUNNEL_COUNT hlog("I2P", "Waiting for Java I2P to establish circuits...") hlog("I2P", f"Check progress at http://127.0.0.1:{I2P_HTTP_PORT}/console") total_wait = 0 check_interval = 10 max_wait = 900 last_status = None last_peers = 0 hlog("I2P", "Waiting for I2P initialization engine...") await asyncio.sleep(15) proc = PROCESSES.get("i2p") if proc and proc.returncode is not None: hlog("I2P_ERR", f"I2P process died unexpectedly with code {proc.returncode}") return False while total_wait < max_wait: total_wait += check_interval proc = PROCESSES.get("i2p") if proc and proc.returncode is not None: hlog("I2P_ERR", f"I2P process died unexpectedly with code {proc.returncode}") return False port_ok = False try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(1.0) result = sock.connect_ex((LOOPBACK_IP, int(I2P_HTTP_PORT))) sock.close() port_ok = (result == 0) except Exception: pass if port_ok: stats = await get_i2p_console_stats() if stats and stats.get("status") != "no_content": status = stats.get("status", "unknown") active_peers = stats.get("active_peers", 0) known_peers = stats.get("known_peers", 0) tunnels = stats.get("tunnels", 0) uptime = stats.get("uptime", "unknown") is_running = status in ["running", "firewalled"] or "shared clients" in str(stats) if status != last_status or active_peers != last_peers: if is_running and active_peers > 0: hlog("I2P_PROGRESS", f"Status: {status}, Active: {active_peers}, Known: {known_peers}, Tunnels: {tunnels}, Uptime: {uptime}") last_status = status last_peers = active_peers await emit_encrypted_state(f"I2P_PROGRESS_{total_wait}") if active_peers >= 3 or (known_peers >= 10 and status in ["running", "firewalled"]): hlog("I2P_DEBUG", "Attempting to get I2P destination via SAM...") I2P_DESTINATION = await get_i2p_destination_via_sam() if I2P_DESTINATION: hlog("I2P", "Got I2P destination via SAM") STATE["i2p"]["destination"] = I2P_DESTINATION STATE["i2p"]["destination_b64"] = I2P_DESTINATION await emit_encrypted_state("I2P_READY_WITH_KEYS") return True else: hlog("I2P_DEBUG", "SAM failed, trying file-based...") hlog("I2P", f"I2P ready after {total_wait}s - Active Peers: {active_peers}, Known: {known_peers}, Tunnels: {tunnels}") I2P_READY = True I2P_NETWORK_OK = True I2P_PEER_COUNT = active_peers I2P_TUNNEL_COUNT = tunnels I2P_DESTINATION = await get_i2p_destination() if not I2P_DESTINATION: hlog("I2P_DEBUG", "File-based mapping failed, trying SAM again...") I2P_DESTINATION = await get_i2p_destination_via_sam() if I2P_DESTINATION: hlog("I2P", "Successfully got I2P destination via SAM bridge") STATE["i2p"]["destination"] = I2P_DESTINATION STATE["i2p"]["destination_b64"] = I2P_DESTINATION await emit_encrypted_state("I2P_READY_WITH_KEYS") else: hlog("I2P_DEBUG", "SAM bridge also failed") if not I2P_DESTINATION: hlog("I2P_ERR", "Critical: Unable to extract valid base64 identity key via any available path.") return False STATE["i2p"]["destination"] = I2P_DESTINATION STATE["i2p"]["destination_b64"] = I2P_DESTINATION STATE["i2p"]["peer_count"] = active_peers STATE["i2p"]["tunnel_count"] = tunnels STATE["i2p"]["tunnels_ready"] = True hlog("I2P_DEBUG", f"Cryptographic identity confirmation: {I2P_DESTINATION[:50]}...") await emit_encrypted_state("I2P_READY_WITH_KEYS") await asyncio.sleep(2) await emit_encrypted_state("I2P_KEYS_PERSISTED") return True if total_wait % 60 == 0: hlog("I2P_STATUS", f"({total_wait}s) {status}: Active: {active_peers}, Known: {known_peers}, Tunnels: {tunnels}") else: if total_wait % 60 == 0: hlog("I2P", f"Console responding but metrics structure uninitialized... ({total_wait}s)") else: if total_wait % 30 == 0: hlog("I2P", f"Waiting for active connection to I2P console port ({I2P_HTTP_PORT})... ({total_wait}s)") await asyncio.sleep(check_interval) hlog("I2P_ERR", f"I2P failed to establish operational circuits within {max_wait}s limit.") return False # ============================================================================ # CLEANUP FUNCTIONS # ============================================================================ def cleanup_wallet_cache(): """Clean up wallet cache directory.""" try: if os.path.exists(WALLET_CACHE_DIR): for f in os.listdir(WALLET_CACHE_DIR): if f.endswith('.lock'): os.remove(os.path.join(WALLET_CACHE_DIR, f)) except Exception: pass def cleanup_i2p(): try: if os.path.exists(I2P_PERSISTENT_DIR): shutil.rmtree(I2P_PERSISTENT_DIR, ignore_errors=True) os.makedirs(I2P_PERSISTENT_DIR, mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/netDb", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/addressbook", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/certificates", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/destinations", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/destinations/private", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/private", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/run", mode=0o755, exist_ok=True) except Exception: pass def cleanup_tor(): try: if os.path.exists(HS_DIR): shutil.rmtree(HS_DIR, ignore_errors=True) os.makedirs(HS_DIR, mode=0o700, exist_ok=True) except Exception: pass atexit.register(cleanup_wallet_cache) atexit.register(cleanup_ringdb) atexit.register(cleanup_i2p) atexit.register(cleanup_tor) def setup_shm_permissions(): try: os.makedirs(I2P_PERSISTENT_DIR, mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/netDb", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/addressbook", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/certificates", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/destinations", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/destinations/private", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/private", mode=0o755, exist_ok=True) os.makedirs(f"{I2P_PERSISTENT_DIR}/run", mode=0o755, exist_ok=True) subprocess.run(["chown", "-R", "i2puser:i2puser", I2P_PERSISTENT_DIR], check=False) subprocess.run(["chmod", "-R", "755", I2P_PERSISTENT_DIR], check=False) os.makedirs(HS_DIR, mode=0o700, exist_ok=True) subprocess.run(["chown", "-R", "toruser:toruser", "/dev/shm/tor"], check=False) subprocess.run(["chmod", "-R", "755", "/dev/shm/tor"], check=False) # Setup RingDB with proper permissions cleanup_ringdb() except Exception: pass def setup_firewall(): hlog("FIREWALL", "Firewall disabled - container network provides isolation") return # ============================================================================ # SYSTEM FUNCTIONS # ============================================================================ async def setup_gpg(): hlog("GPG", f"Importing master key {BOOTSTRAP_ENCRYPTION_KEY}...") keyservers = [ 'hkps://keyserver.ubuntu.com', 'hkps://keys.openpgp.org', 'hkp://keyserver.ubuntu.com', ] imported = False for ks in keyservers: try: res = subprocess.run( [BIN_GPG, '--batch', '--keyserver', ks, '--recv-keys', BOOTSTRAP_ENCRYPTION_KEY], capture_output=True, timeout=15 ) if res.returncode == 0: hlog("GPG", f"Key imported from {ks}") imported = True break except Exception: pass if not imported: try: result = subprocess.run( f"{BIN_CURL} -sL https://pgp.grimore.org/{BOOTSTRAP_ENCRYPTION_KEY}.asc | {BIN_GPG} --batch --import", shell=True, capture_output=True, timeout=15 ) if result.returncode == 0: verify = subprocess.run( [BIN_GPG, '--list-keys', BOOTSTRAP_ENCRYPTION_KEY], capture_output=True, timeout=5 ) if verify.returncode == 0: hlog("GPG", "Key imported via direct download") imported = True except Exception: pass if imported: hlog("GPG", "Master key imported successfully") await emit_encrypted_state("GPG_READY") return True else: hlog("GPG_ERR", "Failed to import master key") await shutdown() sys.exit(1) async def fix_blockchain_permissions(): try: import pwd uid = pwd.getpwnam('monerouser').pw_uid gid = pwd.getpwnam('monerouser').pw_gid subprocess.run(["chown", "-R", f"{uid}:{gid}", DATA_DIR], check=False) subprocess.run(["chmod", "-R", "755", DATA_DIR], check=False) return True except Exception: return False async def debug_binaries(): hlog("DEBUG", "=== BINARY LOCATION DEBUG ===") binaries = [ (BIN_TOR, "Tor"), (BIN_JAVA, "Java"), (BIN_I2PD, "I2Pd"), (BIN_MONEROD, "Monerod"), (BIN_XMR_RPC, "Wallet RPC"), (BIN_XMRIG_PROXY, "XMRig Proxy"), (BIN_GPG, "GPG"), (BIN_CRYPTSETUP, "Cryptsetup"), (BIN_DMSETUP, "Dmsetup"), ] for path, name in binaries: if os.path.exists(path): if os.access(path, os.X_OK): hlog("DEBUG", f"{name}: OK ({path})") else: hlog("DEBUG", f"{name}: exists but not executable ({path})") else: hlog("DEBUG", f"{name}: NOT FOUND ({path})") if os.path.exists(I2P_HOME): hlog("DEBUG", f"I2P Home: OK ({I2P_HOME})") if os.path.exists(f"{I2P_HOME}/lib/router.jar"): hlog("DEBUG", f"I2P router.jar: OK") else: hlog("DEBUG", f"I2P router.jar: NOT FOUND") else: hlog("DEBUG", f"I2P Home: NOT FOUND ({I2P_HOME})") async def run_preflight_checks() -> Tuple[bool, List[str]]: checks_passed = True issues = [] hlog("PREFLIGHT", "Running preflight checks...") binaries = [ (BIN_TOR, "Tor"), (BIN_JAVA, "Java"), (BIN_MONEROD, "Monerod"), (BIN_XMR_RPC, "Wallet RPC"), (BIN_CRYPTSETUP, "Cryptsetup"), (BIN_GPG, "GPG"), (BIN_CURL, "Curl"), (BIN_MOUNT, "Mount"), (BIN_UMOUNT, "Umount"), (BIN_DMSETUP, "Dmsetup"), ] for path, name in binaries: if os.path.exists(path) and os.access(path, os.X_OK): hlog("PREFLIGHT", f"Binary {name} ({path}): OK") else: issues.append(f"Binary {name} ({path}) not found or not executable") checks_passed = False if os.path.exists(I2P_HOME): hlog("PREFLIGHT", f"I2P home {I2P_HOME}: OK") if os.path.exists(f"{I2P_HOME}/lib/router.jar"): hlog("PREFLIGHT", f"I2P router.jar: OK") else: issues.append(f"I2P router.jar not found in {I2P_HOME}/lib/") checks_passed = False else: issues.append(f"I2P home {I2P_HOME} does not exist") checks_passed = False if os.path.exists(BIN_XMRIG_PROXY) and os.access(BIN_XMRIG_PROXY, os.X_OK): hlog("PREFLIGHT", f"XMRig Proxy ({BIN_XMRIG_PROXY}): OK") else: issues.append(f"XMRig Proxy ({BIN_XMRIG_PROXY}) not found or not executable") checks_passed = False try: result = subprocess.run([BIN_DMSETUP, 'status'], capture_output=True, timeout=5) if result.returncode == 0: hlog("PREFLIGHT", "Device mapper: OK") else: issues.append("Device mapper not available") checks_passed = False except Exception as e: issues.append(f"Device mapper error: {e}") checks_passed = False if os.path.exists(LUKS_DEVICE): hlog("PREFLIGHT", f"LUKS device {LUKS_DEVICE}: OK") else: issues.append(f"LUKS device {LUKS_DEVICE} does not exist") checks_passed = False try: import resource soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) if soft >= 8192: hlog("PREFLIGHT", f"File descriptor limit: {soft}/{hard} (OK)") else: issues.append(f"File descriptor limit too low: {soft}/{hard} (need 8192)") checks_passed = False except: pass if os.path.exists("/dev/shm"): hlog("PREFLIGHT", "SHM directory: OK") else: issues.append("/dev/shm does not exist") checks_passed = False await debug_binaries() if checks_passed: hlog("PREFLIGHT", "All critical preflight checks passed!") else: hlog("PREFLIGHT", f"{len(issues)} critical preflight checks failed:") for issue in issues: hlog("PREFLIGHT", f" - {issue}") return checks_passed, issues # ============================================================================ # TOR FUNCTIONS # ============================================================================ async def setup_tor_with_keys(secret_key_b64): global TOR_PRIVATE_KEY cleanup_tor() os.makedirs(HS_DIR, mode=0o700, exist_ok=True) if secret_key_b64: try: key_path = f"{HS_DIR}/hs_ed25519_secret_key" key_bytes = base64.b64decode(secret_key_b64) if key_bytes: with open(key_path, "wb") as f: f.write(key_bytes) subprocess.run(["chown", "toruser:toruser", key_path], check=False) subprocess.run(["chmod", "600", key_path], check=False) TOR_PRIVATE_KEY = secret_key_b64 STATE["tor"]["secret_key_b64"] = secret_key_b64 hlog("TOR", "Restored Tor private key from bootstrap") await emit_encrypted_state("TOR_KEY_RESTORED") else: hlog("TOR_ERR", "Failed to decode Tor key") TOR_PRIVATE_KEY = None except Exception as e: hlog("TOR_ERR", f"Failed to restore Tor key: {e}") TOR_PRIVATE_KEY = None else: hlog("TOR", "No Tor key provided, will generate ephemeral key") TOR_PRIVATE_KEY = None async def wait_for_tor_circuit(): global TOR_READY, TOR_ONION hlog("TOR", "Waiting for Tor to establish circuits...") hostname_ready = False for attempt in range(60): if os.path.exists(f"{HS_DIR}/hostname"): hostname_ready = True with open(f"{HS_DIR}/hostname", "r") as f: TOR_ONION = f.read().strip() STATE["tor"]["onion"] = TOR_ONION hlog("TOR", f"Hostname ready after {attempt+1}s - onion: {TOR_ONION}") await emit_encrypted_state("TOR_HOSTNAME_READY") break if attempt % 10 == 0: hlog("TOR", f"Waiting for hostname... ({attempt}s)") await asyncio.sleep(1) if not hostname_ready: hlog("TOR_ERR", "Hostname not ready after 60s") return False controller = None try: controller = Controller.from_port(port=TOR_CONTROL_PORT) controller.authenticate() hlog("TOR", "Connected to Tor control port") except Exception as e: hlog("TOR_ERR", f"Failed to connect to Tor control port: {e}") return True try: for attempt in range(TOR_MAX_WAIT): try: response = controller.get_info("status/circuit-established") if response == "1": TOR_READY = True hlog("TOR", f"Tor circuit established after {attempt+1}s") await emit_encrypted_state("TOR_CIRCUIT_READY") return True except Exception: pass if attempt % 10 == 0: hlog("TOR", f"Still waiting for Tor circuit... ({attempt+1}s)") await asyncio.sleep(1) finally: if controller: try: controller.close() except Exception: pass hlog("TOR_WARN", f"Tor circuit not ready after {TOR_MAX_WAIT}s, continuing...") return True async def get_tor_peer_count(): try: controller = Controller.from_port(port=TOR_CONTROL_PORT) controller.authenticate() circuits = controller.get_circuits() controller.close() return len(circuits) except Exception: return 0 # ============================================================================ # WALLET RPC FUNCTIONS - FIXED WITH RINGDB AND HEALTH CHECKS # ============================================================================ async def check_wallet_rpc_health(): """Check if wallet RPC is responsive and restart if needed. Only runs if monerod is synced and wallet RPC has been started.""" global LAST_RESTART_TIME, PROCESSES, RUNNING, MONEROD_SYNCED, WALLET_RPC_STARTED if not RUNNING: return # Don't even check if wallet RPC hasn't been started yet if not WALLET_RPC_STARTED: return # Don't restart wallet RPC if monerod isn't synced if not MONEROD_SYNCED: return proc = PROCESSES.get("wallet_rpc") if not proc or proc.returncode is not None: hlog("WALLET_HEALTH", "Wallet RPC process died - restarting") await start_wallet_rpc() return # Check if it's been too long since last restart (cooldown) now = time.time() cooldown = 120 # 2 minutes cooldown between restarts if now - LAST_RESTART_TIME.get("wallet_rpc", 0) < cooldown: return # Check if it's responsive with a short timeout try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.post( f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_languages"}, timeout=5.0 ) if resp.status_code == 200: try: data = resp.json() if "error" not in data: return except json.JSONDecodeError: pass except (httpx.TimeoutException, httpx.ConnectError) as e: hlog("WALLET_HEALTH", f"Wallet RPC not responding: {str(e)[:50]}") hlog("WALLET_HEALTH", "Wallet RPC unresponsive - restarting") try: proc.terminate() await asyncio.sleep(2) if proc.returncode is None: proc.kill() await asyncio.sleep(1) cleanup_wallet_cache() cleanup_ringdb() await start_wallet_rpc() LAST_RESTART_TIME["wallet_rpc"] = now hlog("WALLET_HEALTH", "Wallet RPC restarted successfully") except Exception as e: hlog("WALLET_HEALTH", f"Failed to restart wallet RPC: {e}") except Exception as e: hlog("WALLET_HEALTH", f"Unexpected health check error: {str(e)[:50]}") async def wait_for_wallet_rpc_ready(max_attempts=120, initial_delay=10.0, max_delay=60.0): """ Wait for wallet RPC to become ready with exponential backoff. Args: max_attempts: Maximum number of connection attempts initial_delay: Initial delay in seconds between attempts max_delay: Maximum delay in seconds (capped) """ delay = initial_delay successful_attempts = 0 consecutive_failures = 0 for attempt in range(1, max_attempts + 1): try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.post( f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_languages"}, timeout=5.0 ) if resp.status_code == 200: try: data = resp.json() if "error" not in data: successful_attempts += 1 consecutive_failures = 0 if successful_attempts >= 3: hlog("WALLET", f"Wallet RPC ready after {attempt} attempts") return True delay = max(initial_delay, delay / 1.5) hlog("WALLET_RPC_DEBUG", f"Wallet RPC responding ({successful_attempts}/3 stable checks)") else: hlog("WALLET_RPC_DEBUG", f"Wallet RPC error response: {data.get('error', {}).get('message', 'Unknown')}") consecutive_failures += 1 except json.JSONDecodeError: hlog("WALLET_RPC_DEBUG", "Invalid JSON response from wallet RPC") consecutive_failures += 1 else: hlog("WALLET_RPC_DEBUG", f"Wallet RPC HTTP error: {resp.status_code}") consecutive_failures += 1 except (httpx.TimeoutException, httpx.ConnectError) as e: hlog("WALLET_RPC_DEBUG", f"Wallet RPC connection attempt {attempt}: {str(e)[:50]}") consecutive_failures += 1 except Exception as e: hlog("WALLET_RPC_DEBUG", f"Wallet RPC unexpected error: {str(e)[:50]}") consecutive_failures += 1 if consecutive_failures > 0: backoff = min(delay * (2 ** min(consecutive_failures - 1, 5)), max_delay) jitter = random.uniform(0.8, 1.2) wait_time = backoff * jitter hlog("WALLET_RPC_DEBUG", f"Wallet RPC backoff: waiting {wait_time:.2f}s (attempt {attempt}/{max_attempts})") if attempt % 10 == 0: hlog("WALLET", f"Waiting for wallet RPC... ({attempt}/{max_attempts}) - {consecutive_failures} consecutive failures") await asyncio.sleep(wait_time) else: await asyncio.sleep(0.5) hlog("WALLET_ERR", f"Wallet RPC failed to become ready after {max_attempts} attempts") return False async def get_wallet_balances(max_retries=3): """Get wallet balances with timeout, retry, and health checking.""" for retry in range(max_retries): try: async with httpx.AsyncClient(timeout=5.0) as client: try: test_resp = await client.post( f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_languages"}, timeout=5.0 ) if test_resp.status_code != 200: hlog("WALLET_HEALTH", f"Wallet RPC unresponsive (HTTP {test_resp.status_code})") if retry < max_retries - 1: await asyncio.sleep(2 * (retry + 1)) continue return None, None, {} except (httpx.TimeoutException, httpx.ConnectError) as e: hlog("WALLET_HEALTH", f"Wallet RPC timeout (attempt {retry + 1}/{max_retries}): {str(e)[:50]}") if retry < max_retries - 1: await asyncio.sleep(3 * (retry + 1)) continue return None, None, {} async with httpx.AsyncClient(timeout=5.0) as client: w_resp = await client.post( f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc", json={ "jsonrpc": "2.0", "id": "0", "method": "get_balance", "params": {"account_index": 0} }, timeout=5.0 ) if w_resp.status_code != 200: hlog("BALANCE_ERR", f"Balance HTTP error: {w_resp.status_code}") if retry < max_retries - 1: await asyncio.sleep(2 * (retry + 1)) continue return None, None, {} w_data = w_resp.json().get("result", {}) total_balance = w_data.get("balance", 0) per_sub = w_data.get("per_subaddress", []) primary_balance = 0 derived_balances = {} primary_addr = STATE.get("primary_address", "") derived_addrs = STATE.get("derived_addresses", []) addr_map = {} for entry in derived_addrs: if entry.get("address"): addr_map[entry["address"]] = entry.get("label", "unknown") for sub in per_sub: sub_addr = sub.get("address", "") sub_balance = sub.get("balance", 0) if sub_addr == primary_addr: primary_balance = sub_balance elif sub_addr in addr_map: label = addr_map[sub_addr] derived_balances[label] = sub_balance for entry in derived_addrs: label = entry.get("label", "unknown") addr = entry.get("address", "") if addr == primary_addr and label not in derived_balances: derived_balances[label] = primary_balance primary_xmr = primary_balance / 1e12 total_xmr = total_balance / 1e12 derived_xmr = {} for label, bal in derived_balances.items(): derived_xmr[label] = bal / 1e12 return primary_xmr, total_xmr, derived_xmr except (httpx.TimeoutException, httpx.ConnectError) as e: hlog("BALANCE_ERR", f"Wallet RPC connection error (attempt {retry + 1}/{max_retries}): {str(e)[:50]}") if retry < max_retries - 1: await asyncio.sleep(3 * (retry + 1)) continue return None, None, {} except Exception as e: hlog("BALANCE_ERR", f"Failed to get balances: {str(e)[:100]}") if retry < max_retries - 1: await asyncio.sleep(2 * (retry + 1)) continue return None, None, {} return None, None, {} async def restore_wallet_from_seed(p_seed, r_height, primary_addr, derived_addrs): global STATE, RESTORE_HEIGHT, WALLET_PASSWORD url = f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc" WALLET_PASSWORD = derive_wallet_password(p_seed) hlog("WALLET_RESTORE", f"Restoring wallet with height: {r_height}") hlog("WALLET_RESTORE", f"Seed length: {len(p_seed) if p_seed else 0}") hlog("WALLET_RESTORE", f"Primary address: {primary_addr[:20] if primary_addr else 'None'}...") os.makedirs(WALLET_CACHE_DIR, mode=0o700, exist_ok=True) hlog("WALLET_RESTORE", f"Wallet cache directory: {WALLET_CACHE_DIR}") # Increase timeout to 300 seconds (5 minutes) for wallet restore async with httpx.AsyncClient(timeout=300.0) as client: existing_wallets = [] if os.path.exists(WALLET_CACHE_DIR): for f in os.listdir(WALLET_CACHE_DIR): if f.endswith('.keys'): wallet_name = f.replace('.keys', '') existing_wallets.append(wallet_name) hlog("WALLET_RESTORE", f"Found wallet cache: {wallet_name}") if existing_wallets: for wallet_name in existing_wallets: try: hlog("WALLET_RESTORE", f"Attempting to open existing wallet: {wallet_name}") open_resp = await client.post(url, json={ "jsonrpc": "2.0", "id": "0", "method": "open_wallet", "params": { "filename": wallet_name, "password": WALLET_PASSWORD } }) if open_resp.status_code == 200: result = open_resp.json() if "error" not in result: hlog("WALLET_RESTORE", f"Successfully opened existing wallet: {wallet_name}") addrs_resp = await client.post(url, json={ "jsonrpc": "2.0", "id": "0", "method": "get_address", "params": {"account_index": 0} }) addrs = addrs_resp.json().get('result', {}).get('addresses', []) if addrs: derived_primary = addrs[0].get('address') STATE["primary_address"] = derived_primary hlog("WALLET_RESTORE", f"Primary address from existing wallet: {derived_primary}") await emit_encrypted_state("WALLET_RESTORED") return p_seed, STATE.get("derived_addresses", []) except Exception as e: hlog("WALLET_RESTORE", f"Failed to open {wallet_name}: {e}") hlog("WALLET_RESTORE", "No existing wallet found or couldn't open, restoring from seed...") try: await client.post(url, json={ "jsonrpc": "2.0", "id": "0", "method": "close_wallet" }) await asyncio.sleep(1) except Exception: pass if r_height == 0: r_height = 100000 hlog("WALLET_RESTORE", f"Using default restore height: {r_height}") RESTORE_HEIGHT = r_height wallet_name = f"wallet_{uuid.uuid4().hex[:8]}" hlog("WALLET_RESTORE", f"Restoring wallet: {wallet_name} in {WALLET_CACHE_DIR} with height {r_height}") # This call can take a long time - using 300s timeout restore_result = await client.post(url, json={ "jsonrpc": "2.0", "id": "0", "method": "restore_deterministic_wallet", "params": { "filename": wallet_name, "seed": p_seed, "restore_height": r_height, "language": "English", "password": WALLET_PASSWORD } }) if restore_result.status_code != 200: hlog("WALLET_ERR", f"Restore HTTP error: {restore_result.status_code}") return None, None result_json = restore_result.json() if "error" in result_json: hlog("WALLET_ERR", f"Restore failed: {result_json['error']}") return None, None hlog("WALLET_RESTORE", f"Wallet restored successfully: {wallet_name}") addrs_resp = await client.post(url, json={ "jsonrpc": "2.0", "id": "0", "method": "get_address", "params": {"account_index": 0} }) addrs = addrs_resp.json().get('result', {}).get('addresses', []) if not addrs: hlog("WALLET_ERR", "No addresses found in restored wallet!") return None, None derived_primary = addrs[0].get('address') hlog("WALLET", f"Derived primary address: {derived_primary}") if primary_addr and primary_addr != derived_primary: hlog("WALLET_WARN", f"Provided primary address doesn't match derived, using derived") primary_addr = derived_primary STATE["primary_address"] = derived_primary existing_addresses = [a['address'] for a in addrs] restored_derived = [] for entry in derived_addrs: label = entry.get("label", "unknown") addr = entry.get("address", "") if not addr: continue if addr == derived_primary: restored_derived.append({"label": label, "address": addr}) continue if addr in existing_addresses: restored_derived.append({"label": label, "address": addr}) hlog("WALLET", f"Found existing derived address: {label}") else: hlog("WALLET", f"Creating derived address: {label}") found = False for idx in range(100): create_result = await client.post(url, json={ "jsonrpc": "2.0", "id": "0", "method": "create_address", "params": { "account_index": 0, "label": label, "password": WALLET_PASSWORD } }) if create_result.status_code != 200: hlog("WALLET_WARN", f"create_address HTTP error: {create_result.status_code}") break new_addr = create_result.json().get('result', {}).get('address') if new_addr == addr: found = True restored_derived.append({"label": label, "address": addr}) hlog("WALLET", f"Created derived address: {label}") break if not found: hlog("WALLET_WARN", f"Could not create derived address: {label}") STATE["derived_addresses"] = restored_derived await emit_encrypted_state("WALLET_RESTORED") return p_seed, restored_derived async def start_wallet_rpc(): global PROCESSES, STATE, RUNNING, WALLET_PASSWORD, RESTORE_HEIGHT, WALLET_RPC_STARTED if not RUNNING: return p_seed = STATE.get("recovery", {}).get("seed") r_height = RESTORE_HEIGHT if RESTORE_HEIGHT > 0 else STATE.get("recovery", {}).get("restore_height", 0) primary_addr = STATE.get("primary_address") derived_addrs = STATE.get("derived_addresses", []) if not p_seed: hlog("WALLET_ERR", "No seed available") return # Clean up stale locks and RingDB before starting cleanup_wallet_cache() cleanup_ringdb() setup_ringdb() WALLET_PASSWORD = derive_wallet_password(p_seed) hlog("WALLET", f"Using restore height: {r_height}") os.makedirs(WALLET_CACHE_DIR, mode=0o700, exist_ok=True) cmd = [ BIN_XMR_RPC, "--wallet-dir", WALLET_CACHE_DIR, "--password", WALLET_PASSWORD, "--rpc-bind-port", str(WALLET_RPC_PORT), "--disable-rpc-login", "--daemon-address", f"{LOOPBACK_IP}:{MONEROD_RPC_RESTRICTED_PORT}", "--shared-ringdb-dir", RINGDB_DIR, "--log-level", "1", # Reduced from 3 to improve performance "--trusted-daemon" ] try: hlog("WALLET_CMD", f"Starting wallet RPC with wallet-dir: {WALLET_CACHE_DIR}") hlog("WALLET_CMD", f"RingDB dir: {RINGDB_DIR}") proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) PROCESSES["wallet_rpc"] = proc ready = await wait_for_wallet_rpc_ready(max_attempts=120, initial_delay=10.0, max_delay=60.0) if not ready: hlog("WALLET_HEALTH", "Wallet RPC failed to start within timeout") if proc.returncode is None: proc.terminate() await asyncio.sleep(2) if proc.returncode is None: proc.kill() await asyncio.sleep(1) await shutdown() sys.exit(1) hlog("WALLET", "Restoring/Opening wallet from seed...") s, restored_derived = await restore_wallet_from_seed(p_seed, r_height, primary_addr, derived_addrs) if not s: hlog("WALLET_ERR", "Wallet restoration failed") await shutdown() sys.exit(1) STATE["recovery"]["seed"] = s STATE["recovery"]["restore_height"] = r_height if restored_derived: STATE["derived_addresses"] = restored_derived WALLET_RPC_STARTED = True hlog("WALLET_HEALTH", "Wallet RPC started successfully") except Exception as e: hlog("WALLET_HEALTH", f"Failed to start wallet RPC: {e}") import traceback hlog("WALLET_HEALTH", traceback.format_exc()) await shutdown() sys.exit(1) async def get_primary_address(): try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.post( f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_address", "params": {"account_index": 0}} ) if resp.status_code != 200: return None data = resp.json().get("result", {}) addresses = data.get("addresses", []) if addresses: primary = addresses[0].get("address") if primary: return primary except Exception as e: hlog("GET_ADDR_ERR", f"Failed to get primary address: {e}") return None # ============================================================================ # MONEROD FUNCTIONS # ============================================================================ async def wait_for_monerod_sync(): global MONEROD_SYNCED hlog("MAIN", "Waiting for monerod to sync...") last_height = 0 no_progress_count = 0 max_no_progress = 60 rpc_errors = 0 while RUNNING: try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.post( f"http://{RPC_BIND_IP}:{MONEROD_RPC_MAIN_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_info"}, timeout=10.0 ) data = resp.json().get("result", {}) height = data.get("height", 0) target_height = data.get("target_height", 0) is_synced = data.get("synchronized", False) if is_synced or (target_height > 0 and height >= target_height - 5): MONEROD_SYNCED = True hlog("MAIN", f"Monerod synced! Height: {height}") await emit_encrypted_state("MONEROD_SYNCED") return True if height > last_height: last_height = height no_progress_count = 0 rpc_errors = 0 hlog("MAIN", f"Syncing... {height}/{target_height} (progressing)") else: no_progress_count += 1 if no_progress_count > max_no_progress: hlog("MAIN", f"Sync stalled at height {height}") return False if no_progress_count % 3 == 0: hlog("MAIN", f"Syncing... {height}/{target_height} (no progress)") await asyncio.sleep(10) except Exception as e: rpc_errors += 1 hlog("MAIN", f"RPC error ({rpc_errors}): {str(e)[:50]}") proc = PROCESSES.get("monerod") if proc and proc.returncode is not None: hlog("MAIN", "Monerod process died, aborting") return False if no_progress_count > max_no_progress and rpc_errors > 60: hlog("MAIN", f"Sync stalled with RPC errors ({rpc_errors})") return False await asyncio.sleep(10) return False async def wait_for_block_template(max_retries=BLOCK_TEMPLATE_MAX_RETRIES, retry_delay=BLOCK_TEMPLATE_RETRY_DELAY): global MONEROD_SYNCED, STATE hlog("JOB_MANAGER", "Ensuring wallet RPC is running...") wallet_proc = PROCESSES.get("wallet_rpc") if not wallet_proc or wallet_proc.returncode is not None: await start_wallet_rpc() await asyncio.sleep(3) hlog("JOB_MANAGER", "Waiting for wallet RPC to be ready...") ready = False for attempt in range(20): try: async with httpx.AsyncClient(timeout=2.0) as client: resp = await client.post( f"http://{LOOPBACK_IP}:{WALLET_RPC_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_languages"} ) if resp.status_code == 200: ready = True hlog("JOB_MANAGER", f"Wallet RPC ready after {attempt+1} attempts") break except Exception: pass await asyncio.sleep(1) if not ready: hlog("JOB_MANAGER_ERR", "Wallet RPC not ready after timeout") return None primary_address = STATE.get("primary_address") if not primary_address: hlog("JOB_MANAGER", "Getting primary address from wallet RPC...") primary_address = await get_primary_address() if primary_address: STATE["primary_address"] = primary_address if not primary_address: hlog("JOB_MANAGER_ERR", "No primary wallet address") return None if not MONEROD_SYNCED: synced = await wait_for_monerod_sync() if not synced: hlog("JOB_MANAGER_ERR", "Monerod failed to sync") return None for attempt in range(max_retries): try: async with httpx.AsyncClient(timeout=10.0) as client: template_resp = await client.post( f"http://{RPC_BIND_IP}:{MONEROD_RPC_MAIN_PORT}/json_rpc", json={ "jsonrpc": "2.0", "id": "0", "method": "get_block_template", "params": { "wallet_address": primary_address, "reserve_size": 8 } } ) result = template_resp.json() if "error" in result: error_msg = result["error"].get("message", "Unknown error") if "Core is busy" in error_msg: hlog("JOB_MANAGER", f"Core busy, retrying... (attempt {attempt+1}/{max_retries})") await asyncio.sleep(retry_delay) continue else: hlog("JOB_MANAGER_ERR", f"get_block_template error: {error_msg}") return None template_data = result.get("result", {}) blob = template_data.get("blocktemplate_blob", "") if blob and len(blob) > 0: hlog("JOB_MANAGER", f"Got block template! height: {template_data.get('height')}") await emit_encrypted_state("BLOCK_TEMPLATE_READY") return template_data else: hlog("JOB_MANAGER", f"Empty blob (attempt {attempt+1}/{max_retries})") await asyncio.sleep(retry_delay) except Exception as e: hlog("JOB_MANAGER", f"Exception getting block template: {e}") await asyncio.sleep(retry_delay) hlog("JOB_MANAGER_ERR", "Could not get block template after max retries") return None # ============================================================================ # WEBSOCKET HANDLER - MINING # ============================================================================ async def handle_ws(ws): global ACTIVE_CLIENTS, CLIENT_SESSIONS, RECYCLED_NONCE_RANGES, NEXT_RANGE_MULTIPLIER, CURRENT_JOB_ID ACTIVE_CLIENTS += 1 client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" ALLOWED_RPC_METHODS = {"start_mining", "stop_mining", "mining_status", "pong", "get_info", "submit_share", "get_job"} if RECYCLED_NONCE_RANGES: nonce_start, nonce_end = RECYCLED_NONCE_RANGES.pop(0) hlog("ALLOCATOR", f"Repurposing dropped workload slice: {nonce_start}-{nonce_end} to {client_id}") else: nonce_start = NEXT_RANGE_MULTIPLIER * NONCE_RANGE_MULTIPLIER_BASE nonce_end = (NEXT_RANGE_MULTIPLIER + 1) * NONCE_RANGE_MULTIPLIER_BASE - 1 NEXT_RANGE_MULTIPLIER += 1 hlog("ALLOCATOR", f"Allocating fresh search bounds: {nonce_start}-{nonce_end} to {client_id}") CLIENT_SESSIONS[ws] = { "id": client_id, "is_mining_active": False, "threads_count": 4, "shares": 0, "workload_nonce_start": nonce_start, "workload_nonce_end": nonce_end, "last_seen": time.time(), "ping_id": 0, "client_reported_speed": 0, "current_job_id": None } async def client_ping_loop(): try: while ws in CLIENT_SESSIONS and RUNNING: await asyncio.sleep(15) if time.time() - CLIENT_SESSIONS[ws]["last_seen"] > 25: hlog("KEEPALIVE", f"Client {client_id} missed deadline. Evicting...") break try: CLIENT_SESSIONS[ws]["ping_id"] += 1 await ws.send(json.dumps({"jsonrpc": "2.0", "method": "ping", "id": CLIENT_SESSIONS[ws]["ping_id"]})) except Exception: break except Exception: pass finally: try: await ws.close(code=1001) except Exception: pass ping_task = asyncio.create_task(client_ping_loop()) try: async for m in ws: try: raw = json.loads(m) meth = raw.get("method", "") params = raw.get("params", {}) req_id = raw.get("id", 100) if meth not in ALLOWED_RPC_METHODS: await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "Method not allowed"} })) continue if ws in CLIENT_SESSIONS: CLIENT_SESSIONS[ws]["last_seen"] = time.time() if meth == "pong": p_data = raw.get("params", {}) if ws in CLIENT_SESSIONS: CLIENT_SESSIONS[ws]["client_reported_speed"] = int(p_data.get("speed", 0)) await adjust_difficulty() continue if meth == "submit_share": await handle_share_submission(ws, params, req_id) continue if meth == "get_job": if CURRENT_JOB_ID and CURRENT_JOB_ID in MINING_JOBS: job_data = MINING_JOBS[CURRENT_JOB_ID].copy() if RECYCLED_NONCE_RANGES: nonce_start, nonce_end = RECYCLED_NONCE_RANGES.pop(0) else: nonce_start = NEXT_RANGE_MULTIPLIER * NONCE_RANGE_MULTIPLIER_BASE nonce_end = (NEXT_RANGE_MULTIPLIER + 1) * NONCE_RANGE_MULTIPLIER_BASE - 1 NEXT_RANGE_MULTIPLIER += 1 CLIENT_SESSIONS[ws]["current_job_id"] = CURRENT_JOB_ID CLIENT_SESSIONS[ws]["workload_nonce_start"] = nonce_start CLIENT_SESSIONS[ws]["workload_nonce_end"] = nonce_end job_response = { "jsonrpc": "2.0", "id": req_id, "result": { "job_id": CURRENT_JOB_ID, "blob": job_data.get("blob", ""), "target": job_data.get("target", 0), "difficulty": job_data.get("difficulty", DIFFICULTY), "height": job_data.get("block_height", 0), "seed_hash": job_data.get("seed_hash", ""), "nonce_range_start": nonce_start, "nonce_range_end": nonce_end } } await ws.send(json.dumps(job_response)) else: job_data = await create_mining_job() if job_data is None: return await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "result": { "job_id": job_data["job_id"], "blob": job_data.get("blob", ""), "target": job_data.get("target", 0), "difficulty": job_data.get("difficulty", DIFFICULTY), "height": job_data.get("block_height", 0), "seed_hash": job_data.get("seed_hash", ""), "nonce_range_start": CLIENT_SESSIONS[ws]["workload_nonce_start"], "nonce_range_end": CLIENT_SESSIONS[ws]["workload_nonce_end"] } })) continue if meth == "start_mining": CLIENT_SESSIONS[ws]["is_mining_active"] = True CLIENT_SESSIONS[ws]["threads_count"] = int(params.get("threads_count", 4)) job_data = await create_mining_job() if job_data is None: return CLIENT_SESSIONS[ws]["current_job_id"] = job_data["job_id"] resp = { "jsonrpc": "2.0", "id": req_id, "status": "OK", "untrusted": False, "result": { "status": "OK", "untrusted": False, "workload_id": hashlib.md5(client_id.encode()).hexdigest()[:8], "nonce_range_start": CLIENT_SESSIONS[ws]["workload_nonce_start"], "nonce_range_end": CLIENT_SESSIONS[ws]["workload_nonce_end"], "job_id": job_data["job_id"], "blob": job_data.get("blob", ""), "target": job_data.get("target", 0), "difficulty": job_data.get("difficulty", DIFFICULTY), "height": job_data.get("block_height", 0), "seed_hash": job_data.get("seed_hash", "") } } await ws.send(json.dumps(resp)) hlog("MINING", f"Client {client_id} started mining with job {job_data['job_id']}") continue elif meth == "stop_mining": CLIENT_SESSIONS[ws]["is_mining_active"] = False CLIENT_SESSIONS[ws]["current_job_id"] = None await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "status": "OK", "untrusted": False, "result": {"status": "OK"} })) hlog("MINING", f"Client {client_id} stopped mining") continue elif meth == "mining_status": is_active = CLIENT_SESSIONS[ws]["is_mining_active"] threads = CLIENT_SESSIONS[ws]["threads_count"] speed = int(CLIENT_SESSIONS[ws].get("client_reported_speed", 0)) resp = { "jsonrpc": "2.0", "id": req_id, "result": { "active": is_active, "threads_count": threads, "untrusted": False, "status": "OK", "pow_algorithm": "RandomX", "speed": speed if is_active else 0, "shares": CLIENT_SESSIONS[ws]["shares"], "difficulty": DIFFICULTY } } await ws.send(json.dumps(resp)) continue else: ep = f"http://{RPC_BIND_IP}:{MONEROD_RPC_RESTRICTED_PORT}/json_rpc" if "jsonrpc" not in raw: raw["jsonrpc"] = "2.0" if "id" not in raw: raw["id"] = req_id async with httpx.AsyncClient() as client: resp = (await client.post(ep, json=raw, timeout=30.0)).json() await ws.send(json.dumps(resp)) except Exception: pass except (websockets.exceptions.ConnectionClosed, asyncio.exceptions.IncompleteReadError): pass finally: ping_task.cancel() if ws in CLIENT_SESSIONS: abandoned_start = CLIENT_SESSIONS[ws]["workload_nonce_start"] abandoned_end = CLIENT_SESSIONS[ws]["workload_nonce_end"] RECYCLED_NONCE_RANGES.append((abandoned_start, abandoned_end)) hlog("RECYCLER", f"Client {client_id} closed. Recycled slice: {abandoned_start}-{abandoned_end}") del CLIENT_SESSIONS[ws] ACTIVE_CLIENTS -= 1 # ============================================================================ # MINING FUNCTIONS # ============================================================================ async def create_mining_job(): global CURRENT_JOB_ID, DIFFICULTY job_id = str(uuid.uuid4())[:8] CURRENT_JOB_ID = job_id hlog("JOB_MANAGER", f"Creating mining job {job_id}...") template_data = await wait_for_block_template() if template_data is None: hlog("JOB_MANAGER_ERR", "Could not get block template") return None blob = template_data.get("blocktemplate_blob", "") height = template_data.get("height", 0) difficulty = template_data.get("difficulty", 0) seed_hash = template_data.get("seed_hash", "") target = difficulty if not target and blob: try: blob_bytes = bytes.fromhex(blob) if len(blob_bytes) >= 86: target = int.from_bytes(blob_bytes[78:86], 'little') except Exception: pass job_data = { "job_id": job_id, "block_height": height, "difficulty": DIFFICULTY, "blob": blob, "seed_hash": seed_hash, "target": target, "timestamp": int(time.time()) } MINING_JOBS[job_id] = job_data hlog("JOB_MANAGER", f"Created mining job {job_id} (height: {height})") return job_data async def adjust_difficulty(): global DIFFICULTY, LAST_DIFFICULTY_ADJUST, TOTAL_HASHRATE total_hashrate = 0 active_miners = 0 for session in CLIENT_SESSIONS.values(): if session.get("is_mining_active", False): total_hashrate += session.get("client_reported_speed", 0) active_miners += 1 TOTAL_HASHRATE = total_hashrate if active_miners == 0: DIFFICULTY = DIFFICULTY_INITIAL return try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.post( f"http://{RPC_BIND_IP}:{MONEROD_RPC_MAIN_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_info"} ) data = resp.json().get("result", {}) network_hashrate = data.get("hashrate", 0) if network_hashrate == 0: network_hashrate = data.get("difficulty", 1) / 120 target_shares_per_second = active_miners / 60.0 if target_shares_per_second > 0 and network_hashrate > 0: new_difficulty = max(1000, int(network_hashrate / (total_hashrate + 1) * 100)) else: new_difficulty = DIFFICULTY_INITIAL if new_difficulty > DIFFICULTY * 2: new_difficulty = DIFFICULTY * 2 if new_difficulty < DIFFICULTY / 2: new_difficulty = DIFFICULTY / 2 DIFFICULTY = int(new_difficulty) LAST_DIFFICULTY_ADJUST = time.time() hlog("DIFFICULTY", f"Adjusted to {DIFFICULTY}") except Exception: pass async def handle_share_submission(ws, params, req_id): global DIFFICULTY job_id = params.get("job_id") nonce = params.get("nonce") result = params.get("result") if not all([job_id, nonce, result]): await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32602, "message": "Invalid share: missing fields"} })) return if job_id not in MINING_JOBS: await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": "Invalid job ID"} })) return if ws not in CLIENT_SESSIONS: return session = CLIENT_SESSIONS[ws] try: nonce_val = int(nonce, 16) if isinstance(nonce, str) else int(nonce) if session.get("workload_nonce_start", 0) <= nonce_val <= session.get("workload_nonce_end", 0): async with httpx.AsyncClient(timeout=30.0) as client: submit_resp = await client.post( f"http://{RPC_BIND_IP}:{MONEROD_RPC_RESTRICTED_PORT}/json_rpc", json={ "jsonrpc": "2.0", "id": req_id, "method": "submit_block", "params": {"block": result} } ) response = submit_resp.json() if response.get("status") == "OK": session["shares"] = session.get("shares", 0) + 1 hlog("SHARE", f"Valid share from {session['id']} (nonce: {nonce_val})") await adjust_difficulty() else: hlog("SHARE", f"Invalid share from {session['id']}: {response.get('status', 'unknown')}") await ws.send(json.dumps(response)) else: hlog("SHARE", f"Nonce out of range: {nonce_val}") await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": "Nonce out of range"} })) except (ValueError, TypeError) as e: hlog("SHARE", f"Invalid nonce format: {nonce}") await ws.send(json.dumps({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32602, "message": f"Invalid nonce format: {str(e)}"} })) async def status_heartbeat(): global ACTIVE_CLIENTS, TOTAL_HASHRATE, I2P_PEER_COUNT, I2P_TUNNEL_COUNT, MONEROD_SYNCED while RUNNING: try: await asyncio.sleep(30) await check_wallet_rpc_health() if time.time() - LAST_DIFFICULTY_ADJUST > DIFFICULTY_ADJUST_INTERVAL: await adjust_difficulty() try: i2p_stats = await get_i2p_console_stats() if i2p_stats: I2P_PEER_COUNT = i2p_stats.get('active_peers', 0) I2P_TUNNEL_COUNT = i2p_stats.get('tunnels', 0) except Exception as e: hlog("I2P_STATS_ERR", f"Failed to get I2P stats: {str(e)[:50]}") tor_circuits = 0 try: tor_circuits = await get_tor_peer_count() except Exception as e: hlog("TOR_STATS_ERR", f"Failed to get Tor stats: {str(e)[:50]}") height = 0 target_height = 0 peers_out = 0 peers_in = 0 net_pct = 0.0 try: async with httpx.AsyncClient(timeout=5.0) as client: d_resp = await client.post( f"http://{RPC_BIND_IP}:{MONEROD_RPC_MAIN_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_info"} ) d = d_resp.json().get("result", {}) height = d.get("height", 0) target_height = d.get("target_height", 0) if target_height == 0: target_height = height peers_out = d.get("outgoing_connections_count", 0) peers_in = d.get("incoming_connections_count", 0) net_pct = (height / target_height * 100.0) if target_height > 0 else 0.0 if d.get("synchronized", False): MONEROD_SYNCED = True except Exception as e: hlog("MONEROD_STATS_ERR", f"Failed to get monerod stats: {str(e)[:50]}") primary_bal = None derived_bal = {} if MONEROD_SYNCED: try: primary_bal, total_bal, derived_bal = await get_wallet_balances() except Exception as e: hlog("BALANCE_STATS_ERR", f"Failed to get wallet balances: {str(e)[:50]}") balance_parts = [] if primary_bal is not None: balance_parts.append(f"Primary: {primary_bal:.12f} XMR") for label, bal in derived_bal.items(): if bal is not None: balance_parts.append(f"{label.capitalize()}: {bal:.12f} XMR") for entry in STATE.get("derived_addresses", []): label = entry.get("label", "unknown") if label not in derived_bal: balance_parts.append(f"{label.capitalize()}: 0.000000000000 XMR") balance_str = " | ".join(balance_parts) if balance_parts else "Balances: N/A" total_hashrate = sum(s.get("client_reported_speed", 0) for s in CLIENT_SESSIONS.values()) TOTAL_HASHRATE = total_hashrate active_miners = sum(1 for s in CLIENT_SESSIONS.values() if s.get("is_mining_active", False)) status_parts = [ f"Net: {net_pct:.1f}% ({height}/{target_height})", f"Peers: {peers_out} out / {peers_in} in", f"Tor: {tor_circuits} circuits", f"I2P: {I2P_PEER_COUNT} peers, {I2P_TUNNEL_COUNT} tunnels", f"Clients: {ACTIVE_CLIENTS}", f"Mining: {active_miners} active", f"Hashrate: {total_hashrate:,.0f} H/s", f"Diff: {DIFFICULTY:,}", balance_str ] hlog("STATUS", " | ".join(status_parts)) except Exception as e: hlog("HEARTBEAT_ERR", f"Heartbeat error: {str(e)}") continue async def stream_logs(comp, stream): try: while True: line = await stream.readline() if not line: break clean_line = sanitize_log_line(line.decode().strip()) if "error" in clean_line.lower() or "warn" in clean_line.lower(): hlog(comp, clean_line) continue if comp == "monerod" and is_monerod_spam(clean_line): continue important_patterns = [ "block successfully", "height", "synced", "onion", "destination", "listening", "ready", "started", "stopped", "failed", "error", "warn", "tunnel", "peer", "NTCP2", "SSU2" ] show = False for pattern in important_patterns: if pattern in clean_line.lower(): show = True break if show: hlog(comp, clean_line) except asyncio.CancelledError: pass # ============================================================================ # SERVICE MANAGEMENT # ============================================================================ async def wait_for_services(): global TOR_READY, I2P_READY hlog("MAIN", "Waiting for Tor and I2P to be ready...") tor_task = asyncio.create_task(wait_for_tor_circuit()) i2p_task = asyncio.create_task(wait_for_i2p_circuit()) tor_ready = await tor_task i2p_ready = await i2p_task if not tor_ready: hlog("MAIN_ERR", "Tor failed to start - shutting down") await shutdown() sys.exit(1) if not i2p_ready: hlog("MAIN_ERR", "I2P failed to start - shutting down") await shutdown() sys.exit(1) hlog("MAIN", "Both Tor and I2P are ready") await emit_encrypted_state("ALL_SERVICES_READY") return True async def shutdown(signum=None, frame=None): global RUNNING if not RUNNING: return hlog("SHUTDOWN", "Shutting down orchestrator...") RUNNING = False shutdown_order = ["wallet_rpc", "monerod", "tor", "i2p", "sam"] for name in shutdown_order: proc = PROCESSES.get(name) if proc and hasattr(proc, "returncode") and proc.returncode is None: hlog("SHUTDOWN", f"Terminating {name}...") try: proc.terminate() await asyncio.sleep(0.5) except Exception: pass await asyncio.sleep(2) for name in shutdown_order: proc = PROCESSES.get(name) if proc and hasattr(proc, "returncode") and proc.returncode is None: hlog("SHUTDOWN", f"Killing {name}...") try: proc.kill() except Exception: pass await asyncio.sleep(1) # Emit final complete bootstrap.json before wiping everything await emit_final_state() global WALLET_PASSWORD WALLET_PASSWORD = None cleanup_wallet_cache() cleanup_ringdb() cleanup_i2p() cleanup_tor() if os.path.exists("/dev/shm/bitmonero.log"): os.remove("/dev/shm/bitmonero.log") try: subprocess.run([BIN_UMOUNT, '-l', DATA_DIR], capture_output=True, check=False) except Exception: pass try: subprocess.run([BIN_CRYPTSETUP, 'close', LUKS_NAME], capture_output=True, check=False) except Exception: pass hlog("SHUTDOWN", "Shutdown complete - no persistent keys left") sys.exit(0) # ============================================================================ # BOOTSTRAP FUNCTION # ============================================================================ async def setup_i2p_with_keys(destination_b64, router_keys_b64=None, reseed_b64=None): global I2P_PRIVATE_KEY, I2P_DESTINATION, I2P_ROUTER_KEYS cleanup_i2p() await setup_i2p_directories() await create_i2p_config() if destination_b64: I2P_PRIVATE_KEY = decode_i2p_base64(destination_b64) if I2P_PRIVATE_KEY is None: hlog("I2P_ERR", f"Failed to decode I2P destination_b64, will generate via SAM") I2P_PRIVATE_KEY = None else: STATE["i2p"]["destination_b64"] = destination_b64 hlog("I2P_DEBUG", f"Restored I2P key, length: {len(destination_b64)} chars, decoded length: {len(I2P_PRIVATE_KEY)} bytes") write_paths = [ f"{I2P_PERSISTENT_DIR}/destinations/monerod.dat", f"{I2P_PERSISTENT_DIR}/destinations/private/monerod.dat", f"{I2P_PERSISTENT_DIR}/private/monerod.dat", f"{I2P_CONFIG_DIR}/destinations/monerod.dat", f"{I2P_CONFIG_DIR}/destinations/private/monerod.dat", f"{I2P_CONFIG_DIR}/private/monerod.dat", ] for path in write_paths: try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "wb") as f: f.write(I2P_PRIVATE_KEY) hlog("I2P_DEBUG", f"Wrote I2P private key to: {path}") except Exception as e: hlog("I2P_WARN", f"Could not write key to {path}: {e}") hlog("I2P", "Restored I2P private key from bootstrap") await emit_encrypted_state("I2P_KEY_RESTORED") else: hlog("I2P", "No I2P key provided, will generate via SAM") if router_keys_b64: STATE["i2p"]["router_keys_b64"] = router_keys_b64 try: I2P_ROUTER_KEYS = base64.b64decode(router_keys_b64) with open(f"{I2P_PERSISTENT_DIR}/router_keys.dat", "wb") as f: f.write(I2P_ROUTER_KEYS) except Exception as e: hlog("I2P_WARN", f"Could not write router keys: {e}") if reseed_b64: STATE["i2p"]["reseed_su3_b64"] = reseed_b64 try: reseed_data = base64.b64decode(reseed_b64) with open(f"{I2P_PERSISTENT_DIR}/reseed.su3", "wb") as f: f.write(reseed_data) except Exception as e: hlog("I2P_WARN", f"Could not write reseed data: {e}") async def bootstrap(): global RESTORE_HEIGHT await setup_gpg() setup_shm_permissions() setup_firewall() try: raw_input = sys.stdin.read().strip() if not raw_input: hlog("BOOTSTRAP_ERR", "No input received on stdin!") await shutdown() sys.exit(1) payload = json.loads(raw_input) luks_key_b64 = payload.get('luks_key') if not luks_key_b64: hlog("LUKS_ERR", "No LUKS key provided") await shutdown() sys.exit(1) STATE["luks_key"] = luks_key_b64 tor_key = payload.get("tor", {}).get("secret_key_b64") await setup_tor_with_keys(tor_key) i2p_data = payload.get("i2p", {}) i2p_key = i2p_data.get("destination_b64") router_keys_b64 = i2p_data.get("router_keys_b64") reseed_b64 = i2p_data.get("reseed_su3_b64") if i2p_key: hlog("I2P_BOOTSTRAP", f"Restoring I2P key from payload") else: hlog("I2P_BOOTSTRAP", "No I2P key in payload - will generate via SAM") await setup_i2p_with_keys(i2p_key, router_keys_b64, reseed_b64) hlog("LUKS", "Decoding LUKS key...") luks_key = base64.b64decode(luks_key_b64) if luks_key is None: hlog("LUKS_ERR", "Failed to decode LUKS key") await shutdown() sys.exit(1) hlog("LUKS", f"Closing any existing {LUKS_NAME} device...") subprocess.run([BIN_CRYPTSETUP, 'close', LUKS_NAME], capture_output=True) hlog("LUKS", f"Opening LUKS device {LUKS_DEVICE} as {LUKS_NAME}...") unlock = subprocess.Popen( [BIN_CRYPTSETUP, 'open', LUKS_DEVICE, LUKS_NAME, '--type', LUKS_TYPE, '--cipher', LUKS_CIPHER, '--key-size', LUKS_KEY_SIZE, '--hash', LUKS_HASH, '--key-file', '-'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout, stderr = unlock.communicate(input=luks_key) luks_key = None gc.collect() if unlock.returncode != 0: hlog("LUKS_ERR", f"Failed to open LUKS device: {stderr.decode().strip() if stderr else 'Unknown error'}") await shutdown() sys.exit(1) hlog("LUKS", "LUKS device opened successfully") await emit_encrypted_state("LUKS_OPENED") subprocess.run([BIN_DMSETUP, 'mknodes', LUKS_NAME], env=dict(os.environ, DM_DISABLE_UDEV="1"), capture_output=True) os.makedirs(DATA_DIR, exist_ok=True) mount_result = subprocess.run([BIN_MOUNT, f'/dev/mapper/{LUKS_NAME}', DATA_DIR], capture_output=True) if mount_result.returncode != 0: mount_result = subprocess.run([BIN_MOUNT, '-t', 'ext4', f'/dev/mapper/{LUKS_NAME}', DATA_DIR], capture_output=True) if mount_result.returncode != 0: hlog("LUKS_ERR", f"Mount failed: {mount_result.stderr.decode().strip() if mount_result.stderr else 'Unknown error'}") await shutdown() sys.exit(1) hlog("LUKS", "Successfully mounted encrypted volume") await emit_encrypted_state("LUKS_MOUNTED") await fix_blockchain_permissions() seed = payload.get("recovery", {}).get("seed") restore_height = payload.get("recovery", {}).get("restore_height", 0) primary_addr = payload.get("primary_address", "") derived_addrs = payload.get("derived_addresses", []) RESTORE_HEIGHT = restore_height hlog("BOOTSTRAP", f"Restore height from bootstrap: {restore_height}") hlog("BOOTSTRAP", f"Seed present: {bool(seed)}") hlog("BOOTSTRAP", f"Primary address: {primary_addr[:20] if primary_addr else 'None'}...") hlog("BOOTSTRAP", f"Derived addresses: {len(derived_addrs)}") if not seed: hlog("WALLET_WARN", "No seed provided") await shutdown() sys.exit(1) STATE["recovery"]["seed"] = seed STATE["recovery"]["restore_height"] = restore_height STATE["primary_address"] = primary_addr STATE["derived_addresses"] = derived_addrs await emit_encrypted_state("BOOTSTRAP_COMPLETE") return payload except Exception as e: hlog("BOOTSTRAP_ERR", f"Bootstrap failed: {e}") print_raw(generate_bf_log({"critical": traceback.format_exc()})) await shutdown() sys.exit(1) # ============================================================================ # MAIN FUNCTION - FIXED WITH LINEAR STARTUP ORDER AND IMPROVED STABILITY # ============================================================================ async def main(): global RESTORE_HEIGHT, MONEROD_SYNCED, WALLET_RPC_STARTED, MONEROD_RPC_READY # Initialize flags WALLET_RPC_STARTED = False MONEROD_SYNCED = False MONEROD_RPC_READY = False try: self_bytes = Path(__file__).read_bytes() sha256_digest = hashlib.sha256(self_bytes).hexdigest() print_raw(f"\n[INTEGRITY] RUNNING SCRIPT SHA-256 DIGEST: {sha256_digest}\n") except Exception: pass try: import resource resource.setrlimit(resource.RLIMIT_NOFILE, (8192, 8192)) hlog("SYSTEM", "File descriptor limit raised to 8192") except Exception: pass loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): try: loop.add_signal_handler(sig, lambda s=sig: asyncio.create_task(shutdown(s))) except NotImplementedError: signal.signal(sig, lambda signum, frame: asyncio.run_coroutine_threadsafe(shutdown(signum, frame), loop)) # ======================================================================== # STEP 1: BOOTSTRAP # ======================================================================== payload = await bootstrap() # ======================================================================== # STEP 2: PREFLIGHT CHECKS # ======================================================================== checks_passed, issues = await run_preflight_checks() await emit_encrypted_state("PREFLIGHT_COMPLETE") if not checks_passed: hlog("MAIN_ERR", f"Preflight checks failed: {len(issues)} issues found") for issue in issues: hlog("MAIN_ERR", f" - {issue}") await shutdown() sys.exit(1) # ======================================================================== # STEP 3: START TOR # ======================================================================== hlog("MAIN", "Starting Tor...") tor_data_dir = "/dev/shm/tor" os.makedirs(tor_data_dir, exist_ok=True, mode=0o700) os.makedirs(HS_DIR, exist_ok=True, mode=0o700) try: subprocess.run(["chown", "-R", "toruser:toruser", tor_data_dir], check=True) subprocess.run(["chmod", "700", tor_data_dir], check=True) tor_user = "toruser" except Exception: tor_user = "root" tor_cmd = [ BIN_TOR, "--User", tor_user, "--SocksPort", f"{LOOPBACK_IP}:{TOR_SOCKS_PORT}", "--ControlPort", str(TOR_CONTROL_PORT), "--CookieAuthentication", "0", "--DataDirectory", tor_data_dir, "--HiddenServiceDir", HS_DIR, "--HiddenServiceVersion", "3", "--HiddenServicePort", f"80 {LOOPBACK_IP}:{WS_INTERNAL_PORT}", "--HiddenServicePort", f"{P2P_EXT_PORT} {LOOPBACK_IP}:{MONEROD_P2P_PORT}", "--CircuitBuildTimeout", "60", "--KeepalivePeriod", "600", "--NumEntryGuards", "8" ] t_proc = await asyncio.create_subprocess_exec(*tor_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT) PROCESSES["tor"] = t_proc TASKS.append(asyncio.create_task(stream_logs("tor", t_proc.stdout))) await emit_encrypted_state("TOR_STARTED") # ======================================================================== # STEP 4: START I2P # ======================================================================== hlog("MAIN", "Starting Java I2P...") await setup_i2p_directories() await create_i2p_config() i2p_lib_dir = f"{I2P_HOME}/lib" i2p_jars = [] if os.path.exists(i2p_lib_dir): for f in os.listdir(i2p_lib_dir): if f.endswith('.jar'): i2p_jars.append(f"{i2p_lib_dir}/{f}") classpath = ":".join(i2p_jars) hlog("I2P", f"Classpath has {len(i2p_jars)} JARs") i2p_env = os.environ.copy() i2p_env["HOME"] = "/var/lib/i2p" i2p_env["USER"] = "i2puser" i2p_env["LOGNAME"] = "i2puser" i2p_env["I2P"] = I2P_HOME i2p_env["I2P_DATA"] = I2P_DATA_DIR i2p_env["I2P_CONFIG_DIR"] = I2P_CONFIG_DIR i2p_env["JAVA_OPTS"] = "-Xmx512m -Xms128m" i2p_cmd = [ BIN_SUDO, "-u", "i2puser", BIN_JAVA, "-cp", classpath, f"-Di2p.config.dir={I2P_CONFIG_DIR}", f"-Di2p.data.dir={I2P_DATA_DIR}", "net.i2p.router.RouterLaunch" ] hlog("I2P_CMD", f"Starting Java I2P with: {' '.join(i2p_cmd)}") await emit_encrypted_state("I2P_STARTING") i2p_proc = await asyncio.create_subprocess_exec( *i2p_cmd, env=i2p_env, cwd=I2P_HOME, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT ) PROCESSES["i2p"] = i2p_proc TASKS.append(asyncio.create_task(stream_i2p_logs(i2p_proc.stdout))) await emit_encrypted_state("I2P_RUNNING") # ======================================================================== # STEP 5: WAIT FOR TOR AND I2P TO BE READY # ======================================================================== services_ready = await wait_for_services() if not services_ready: hlog("MAIN_ERR", "Services not ready - shutting down") await shutdown() sys.exit(1) # ======================================================================== # STEP 6: START MONEROD # ======================================================================== hlog("MAIN", "All services ready, starting Monerod...") port_available = False for retry in range(10): if is_port_available(LOOPBACK_IP, int(MONEROD_P2P_PORT)): port_available = True hlog("MAIN", f"Port {MONEROD_P2P_PORT} is available (attempt {retry+1}/10)") break await asyncio.sleep(5) if not port_available: hlog("MAIN_ERR", f"FATAL: Port {MONEROD_P2P_PORT} is in use") await shutdown() sys.exit(1) onion = None try: with open(f"{HS_DIR}/hostname", "r") as f: onion = f.read().strip() TOR_ONION = onion STATE["tor"]["onion"] = onion hlog("TOR", f"Onion Identity: {onion}") except FileNotFoundError: hlog("TOR_WARN", "Hostname file not found") i2p_dest = await get_i2p_destination_via_sam() if i2p_dest: hlog("I2P", f"I2P Destination (SAM): {i2p_dest[:20]}...") else: hlog("I2P_WARN", "I2P destination not found - will try again later") m_cmd = [ BIN_SUDO, "-u", "monerouser", BIN_MONEROD, "--non-interactive", "--data-dir", DATA_DIR, "--rpc-bind-ip", RPC_BIND_IP, "--rpc-bind-port", str(MONEROD_RPC_MAIN_PORT), "--rpc-restricted-bind-ip", RPC_BIND_IP, "--rpc-restricted-bind-port", str(MONEROD_RPC_RESTRICTED_PORT), "--zmq-rpc-bind-ip", "127.0.0.1", "--zmq-rpc-bind-port", "18084", "--p2p-bind-ip", P2P_BIND_IP, "--p2p-bind-port", str(MONEROD_P2P_PORT), "--p2p-ignore-ipv4", "--allow-local-ip", "--pad-transactions", "--enable-dns-blocklist", "--proxy", f"{LOOPBACK_IP}:{TOR_SOCKS_PORT}", "--bootstrap-daemon-proxy", f"{LOOPBACK_IP}:{TOR_SOCKS_PORT}", "--tx-proxy", f"tor,{LOOPBACK_IP}:{TOR_SOCKS_PORT},16", "--tx-proxy", f"i2p,{LOOPBACK_IP}:{I2P_SAM_PORT},1", "--anonymous-inbound", f"{onion if onion else 'localhost'}:{P2P_EXT_PORT},{P2P_BIND_IP}:{MONEROD_P2P_INTERNAL_PORT},25", "--confirm-external-bind", "--log-level", "1", "--max-log-file-size", "0", "--log-file", "/dev/null", "--rpc-payment-allow-free-loopback", "--no-igd", "--p2p-external-port", str(P2P_EXT_PORT), ] for p in ONION_PEERS: m_cmd.extend(["--add-peer", p]) m_proc = await asyncio.create_subprocess_exec(*m_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT) PROCESSES["monerod"] = m_proc TASKS.append(asyncio.create_task(stream_logs("monerod", m_proc.stdout))) await emit_encrypted_state("MONEROD_STARTED") # ======================================================================== # STEP 7: WAIT FOR MONEROD RPC TO BE READY # ======================================================================== hlog("MAIN", "Waiting for monerod RPC...") rpc_ready = False for i in range(RPC_READY_MAX_ATTEMPTS): try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.post( f"http://{LOOPBACK_IP}:{MONEROD_RPC_MAIN_PORT}/json_rpc", json={"jsonrpc": "2.0", "id": "0", "method": "get_info"}, timeout=5.0 ) if resp.status_code == 200: rpc_ready = True MONEROD_RPC_READY = True hlog("MAIN", f"Monerod RPC ready after {i+1} seconds") await emit_encrypted_state("MONEROD_RPC_READY") break except Exception as e: if i % 10 == 0: hlog("MAIN", f"Waiting for monerod RPC... ({i+1}/{RPC_READY_MAX_ATTEMPTS})") await asyncio.sleep(1) if not rpc_ready: hlog("MAIN_WARN", "Monerod RPC not ready - continuing anyway") # ======================================================================== # STEP 8: WAIT FOR MONEROD SYNC (BEFORE STARTING WALLET) # ======================================================================== hlog("MAIN", "Waiting for monerod to sync before starting wallet...") synced = False for attempt in range(SYNC_MAX_ATTEMPTS): if await is_monerod_synced(): synced = True MONEROD_SYNCED = True hlog("MAIN", f"Monerod synced after {attempt+1} attempts") await emit_encrypted_state("MONEROD_SYNCED") break if attempt % 10 == 0: hlog("MAIN", f"Waiting for monerod sync... ({attempt+1}/{SYNC_MAX_ATTEMPTS})") await asyncio.sleep(10) if not synced: hlog("MAIN_ERR", "Monerod failed to sync - shutting down") await shutdown() sys.exit(1) # ======================================================================== # STEP 9: START WALLET RPC (ONLY AFTER MONEROD IS SYNCED) # ======================================================================== hlog("MAIN", "Starting wallet RPC...") await start_wallet_rpc() # WALLET_RPC_STARTED is set inside start_wallet_rpc() # ======================================================================== # STEP 10: START HEARTBEAT (NOW THAT WALLET RPC IS RUNNING) # ======================================================================== TASKS.append(asyncio.create_task(status_heartbeat())) # ======================================================================== # STEP 11: GET BLOCK TEMPLATE # ======================================================================== hlog("MAIN", "Waiting for block template...") template_data = await wait_for_block_template(max_retries=60, retry_delay=10) if template_data is None: hlog("MAIN_ERR", "Could not get block template - shutting down") await shutdown() return blob = template_data.get("blocktemplate_blob", "") height = template_data.get("height", 0) difficulty = template_data.get("difficulty", 0) seed_hash = template_data.get("seed_hash", "") target = difficulty if not target and blob: try: blob_bytes = bytes.fromhex(blob) if len(blob_bytes) >= 86: target = int.from_bytes(blob_bytes[78:86], 'little') except Exception: pass job_id = str(uuid.uuid4())[:8] CURRENT_JOB_ID = job_id job_data = { "job_id": job_id, "block_height": height, "difficulty": DIFFICULTY, "blob": blob, "seed_hash": seed_hash, "target": target, "timestamp": int(time.time()) } MINING_JOBS[job_id] = job_data await adjust_difficulty() # ======================================================================== # STEP 12: SYSTEM ONLINE # ======================================================================== hlog("MAIN", "--- SYSTEMS ONLINE ---") hlog("MAIN", f"Restore Height: {RESTORE_HEIGHT}") print_raw(f"\nPRIMARY ADDR: {STATE['primary_address']}") for entry in STATE.get("derived_addresses", []): print_raw(f"{entry.get('label', 'derived').upper()} ADDR: {entry.get('address')}") print_raw(f"\nTOR ONION: {onion if onion else 'Not available'}") i2p_display = I2P_DESTINATION if I2P_DESTINATION else "Not available" if I2P_DESTINATION: i2p_display = I2P_DESTINATION[:30] + "..." print_raw(f"I2P DEST: {i2p_display}") print_raw(f"I2P KEY CAPTURED: {STATE['i2p']['destination_b64'] is not None}") print_raw(f"\nRPC Main Port: {MONEROD_RPC_MAIN_PORT} on {RPC_BIND_IP}") print_raw(f"RPC Restricted Port: {MONEROD_RPC_RESTRICTED_PORT} on {RPC_BIND_IP}") print_raw(f"Wallet RPC Port: {WALLET_RPC_PORT} on {LOOPBACK_IP}") print_raw(f"I2P HTTP Console: {I2P_HTTP_PORT} on {LOOPBACK_IP}") print_raw(f"Monerod P2P Port: {MONEROD_P2P_PORT} on {P2P_BIND_IP}\n") await emit_encrypted_state("SYSTEMS_ONLINE") # ======================================================================== # STEP 13: MAIN LOOP # ======================================================================== async with websockets.serve(handle_ws, LOOPBACK_IP, WS_INTERNAL_PORT): while RUNNING: # Only check critical services - wallet RPC is handled by health check for name, proc in [("tor", t_proc), ("monerod", m_proc), ("i2p", i2p_proc)]: if proc and proc.returncode is not None: hlog("MAIN_ERR", f"{name} died with code {proc.returncode}") await shutdown() sys.exit(1) # Don't shut down if wallet RPC dies - the health check will restart it wallet_proc = PROCESSES.get("wallet_rpc") if wallet_proc and wallet_proc.returncode is not None: hlog("WALLET_HEALTH", "Wallet RPC process died - health check will restart") # Force an immediate health check await check_wallet_rpc_health() if int(time.time()) % 300 < 5: if not STATE["i2p"]["destination_b64"]: hlog("I2P_DEBUG", "Periodic check: I2P key missing, attempting via SAM...") await get_i2p_destination_via_sam() if STATE["i2p"]["destination_b64"]: hlog("I2P", "Periodic check: Got I2P destination via SAM") await emit_encrypted_state("PERIODIC_STATE") await asyncio.sleep(5) await shutdown() if __name__ == "__main__": asyncio.run(main())