mmgen-wallet/mmgen/proto/xmr/daemon.py

#!/usr/bin/env python3
#
# mmgen = Multi-Mode GENerator, a command-line cryptocurrency wallet
# Copyright (C)2013-2024 The MMGen Project <mmgen@tuta.io>
# Licensed under the GNU General Public License, Version 3:
#   https://www.gnu.org/licenses
# Public project repositories:
#   https://github.com/mmgen/mmgen-wallet
#   https://gitlab.com/mmgen/mmgen-wallet

"""
proto.xmr.daemon: Monero base protocol daemon classes
"""

import os

from ...cfg import gc
from ...util import list_gen,die,contains_any
from ...daemon import CoinDaemon,RPCDaemon,_nw,_dd

class monero_daemon(CoinDaemon):
	daemon_data = _dd('Monero', 18003003, '0.18.3.3-release')
	networks = ('mainnet','testnet')
	exec_fn = 'monerod'
	testnet_dir = 'stagenet'
	new_console_mswin = True
	rpc_ports = _nw(18081, 38081, None) # testnet is stagenet
	cfg_file = 'bitmonero.conf'
	datadirs = {
		'linux': [gc.home_dir,'.bitmonero'],
		'win32': ['/','c','ProgramData','bitmonero']
	}

	def init_datadir(self):
		self.logdir = super().init_datadir()
		return os.path.join(
			self.logdir,
			self.testnet_dir if self.network == 'testnet' else '' )

	def get_p2p_port(self):
		return self.rpc_port - 1

	def init_subclass(self):

		from .rpc import MoneroRPCClient
		self.rpc = MoneroRPCClient(
			cfg    = self.cfg,
			proto  = self.proto,
			host   = 'localhost',
			port   = self.rpc_port,
			user   = None,
			passwd = None,
			test_connection = False,
			daemon = self )

		self.shared_args = list_gen(
			['--no-zmq'],
			[f'--p2p-bind-port={self.p2p_port}', self.p2p_port],
			[f'--rpc-bind-port={self.rpc_port}'],
			['--stagenet', self.network == 'testnet'],
		)

		self.coind_args = list_gen(
			['--hide-my-port'],
			['--no-igd'],
			[f'--data-dir={self.datadir}', self.non_dfl_datadir],
			[f'--pidfile={self.pidfile}', self.platform == 'linux'],
			['--detach',                  not (self.opt.no_daemonize or self.platform=='win32')],
			['--offline',                 not self.opt.online],
		)

	@property
	def stop_cmd(self):
		if self.platform == 'win32':
			return ['kill','-Wf',self.pid]
		elif contains_any( self.start_cmd, ['--restricted-rpc','--public-node'] ):
			return ['kill',self.pid]
		else:
			return [self.exec_fn] + self.shared_args + ['exit']

class MoneroWalletDaemon(RPCDaemon):

	master_daemon = 'monero_daemon'
	rpc_type = 'Monero wallet'
	exec_fn = 'monero-wallet-rpc'
	coin = 'XMR'
	new_console_mswin = True
	networks = ('mainnet','testnet')
	rpc_ports = _nw(13131, 13141, None) # testnet is non-standard
	_reset_ok = ('debug','wait','pids','force_kill')
	test_suite_datadir = os.path.join('test','daemons','xmrtest','wallet_rpc')

	def start(self,*args,**kwargs):
		try: # NB: required due to bug in v18.3.1: PID file not deleted on shutdown
			os.unlink(self.pidfile)
		except FileNotFoundError:
			pass
		super().start(*args,**kwargs)

	def __init__(
			self,
			cfg,
			proto,
			wallet_dir  = None,
			test_suite  = False,
			user        = None,
			passwd      = None,
			monerod_addr = None,
			proxy       = None,
			port_shift  = None,
			datadir     = None,
			trust_monerod = False,
			test_monerod = False,
			opts         = None,
			flags        = None ):

		self.proto = proto
		self.test_suite = test_suite

		super().__init__(cfg,opts=opts,flags=flags)

		self.network = proto.network
		self.wallet_dir = wallet_dir or (self.test_suite_datadir if test_suite else None)
		self.rpc_port = (
			self.cfg.wallet_rpc_port or
			getattr(self.rpc_ports,self.network) + (11 if test_suite else 0) )
		if port_shift:
			self.rpc_port += port_shift

		id_str = f'{self.exec_fn}-{self.bind_port}'
		self.datadir = datadir or (self.test_suite_datadir if test_suite else self.exec_fn + '.d')
		self.pidfile = os.path.join(self.datadir,id_str+'.pid')
		self.logfile = os.path.join(self.datadir,id_str+'.log')

		self.proxy = proxy
		self.monerod_addr = monerod_addr
		self.monerod_port = (
			None if monerod_addr else
			CoinDaemon(
				cfg        = self.cfg,
				proto      = proto,
				test_suite = test_suite).rpc_port
		)

		if test_monerod and self.monerod_port:
			import socket
			try:
				socket.create_connection(('localhost',self.monerod_port),timeout=1).close()
			except:
				die( 'SocketError', f'Unable to connect to Monero daemon at localhost:{self.monerod_port}' )

		self.user = user or self.cfg.wallet_rpc_user or self.cfg.monero_wallet_rpc_user
		self.passwd = passwd or self.cfg.wallet_rpc_password or self.cfg.monero_wallet_rpc_password

		assert self.user
		if not self.passwd:
			die(1,
				'You must set your Monero wallet RPC password.\n' +
				'This can be done on the command line with the --wallet-rpc-password option\n' +
				"(insecure, not recommended), or by setting 'monero_wallet_rpc_password' in\n" +
				"the MMGen config file." )

		self.daemon_args = list_gen(
			['--trusted-daemon', trust_monerod],
			['--untrusted-daemon', not trust_monerod],
			[f'--rpc-bind-port={self.rpc_port}'],
			[f'--wallet-dir={self.wallet_dir}'],
			[f'--log-file={self.logfile}'],
			[f'--rpc-login={self.user}:{self.passwd}'],
			[f'--daemon-address={self.monerod_addr}', self.monerod_addr],
			[f'--daemon-port={self.monerod_port}',    not self.monerod_addr],
			[f'--proxy={self.proxy}',                self.proxy],
			[f'--pidfile={self.pidfile}',            self.platform == 'linux'],
			['--detach',                             not (self.opt.no_daemonize or self.platform=='win32')],
			['--stagenet',                           self.network == 'testnet'],
			['--allow-mismatched-daemon-version',    test_suite],
		)

		from .rpc import MoneroWalletRPCClient
		self.rpc = MoneroWalletRPCClient(
			cfg             = self.cfg,
			daemon          = self,
			test_connection = False )