rpc.py: move set_auth() to `proto.btc`; add bad auth test

mmgen/data/version

@@ -1 +1 @@
-13.3.dev5
+13.3.dev6

mmgen/proto/btc/rpc.py

@@ -16,9 +16,33 @@ import os
 
 from ...globalvars import g
 from ...base_obj import AsyncInit
-from ...util import ymsg,vmsg,die
+from ...util import ymsg,vmsg,die,fmt
 from ...fileutil import get_lines_from_file
-from ...rpc import RPCClient
+from ...rpc import RPCClient,auth_data
+
+no_credentials_errmsg = """
+	Error: no {proto_name} RPC authentication method found
+
+	RPC credentials must be supplied using one of the following methods:
+
+	1) If daemon is local and running as same user as you:
+
+	   - no credentials required, or matching rpcuser/rpcpassword and
+	     rpc_user/rpc_password values in {cf_name}.conf and mmgen.cfg
+
+	2) If daemon is running remotely or as different user:
+
+	   - matching credentials in {cf_name}.conf and mmgen.cfg as described
+	     above
+
+	The --rpc-user/--rpc-password options may be supplied on the MMGen command
+	line.  They override the corresponding values in mmgen.cfg. Set them to an
+	empty string to use cookie authentication with a local server when the
+	options are set in mmgen.cfg.
+
+	For better security, rpcauth should be used in {cf_name}.conf instead of
+	rpcuser/rpcpassword.
+"""
 
 class CallSigs:
 
@@ -156,6 +180,33 @@ class BitcoinRPCClient(RPCClient,metaclass=AsyncInit):
 		if g.regtest_user:
 			self.wallet_path = f'/wallet/{g.regtest_user}'
 
+	def set_auth(self):
+		"""
+		MMGen's credentials override coin daemon's
+		"""
+		if g.rpc_user:
+			user,passwd = (g.rpc_user,g.rpc_password)
+		else:
+			user,passwd = self.get_daemon_cfg_options(('rpcuser','rpcpassword')).values()
+
+		if not (user and passwd):
+			user,passwd = (self.daemon.rpc_user,self.daemon.rpc_password)
+
+		if user and passwd:
+			self.auth = auth_data(user,passwd)
+			return
+
+		if self.has_auth_cookie:
+			cookie = self.get_daemon_auth_cookie()
+			if cookie:
+				self.auth = auth_data(*cookie.split(':'))
+				return
+
+		die(1, '\n\n' + fmt(no_credentials_errmsg,strip_char='\t',indent='  ').format(
+				proto_name = self.proto.name,
+				cf_name = (self.proto.is_fork_of or self.proto.name).lower(),
+			))
+
 	def make_host_path(self,wallet):
 		return f'/wallet/{wallet}' if wallet else self.wallet_path
 

mmgen/rpc.py

@@ -30,30 +30,6 @@ from .objmethods import Hilite,InitErrors
 
 auth_data = namedtuple('rpc_auth_data',['user','passwd'])
 
-rpc_credentials_msg = '\n'+fmt("""
-	Error: no {proto_name} RPC authentication method found
-
-	RPC credentials must be supplied using one of the following methods:
-
-	A) If daemon is local and running as same user as you:
-
-	   - no credentials required, or matching rpcuser/rpcpassword and
-	     rpc_user/rpc_password values in {cf_name}.conf and mmgen.cfg
-
-	B) If daemon is running remotely or as different user:
-
-	   - matching credentials in {cf_name}.conf and mmgen.cfg as described above
-
-	The --rpc-user/--rpc-password options may be supplied on the MMGen command line.
-	They override the corresponding values in mmgen.cfg. Set them to an empty string
-	to use cookie authentication with a local server when the options are set
-	in mmgen.cfg.
-
-	For better security, rpcauth should be used in {cf_name}.conf instead of
-	rpcuser/rpcpassword.
-
-""",strip_char='\t')
-
 def dmsg_rpc(fs,data=None,is_json=False):
 	if g.debug_rpc:
 		msg(
@@ -318,33 +294,6 @@ class RPCClient(MMGenObject):
 		ret = self._get_backend(backend)
 		self.backend = (await ret) if type(ret).__name__ == 'coroutine' else ret
 
-	def set_auth(self):
-		"""
-		MMGen's credentials override coin daemon's
-		"""
-		if g.rpc_user:
-			user,passwd = (g.rpc_user,g.rpc_password)
-		else:
-			user,passwd = self.get_daemon_cfg_options(('rpcuser','rpcpassword')).values()
-
-		if not (user and passwd):
-			user,passwd = (self.daemon.rpc_user,self.daemon.rpc_password)
-
-		if user and passwd:
-			self.auth = auth_data(user,passwd)
-			return
-
-		if self.has_auth_cookie:
-			cookie = self.get_daemon_auth_cookie()
-			if cookie:
-				self.auth = auth_data(*cookie.split(':'))
-				return
-
-		die(1,rpc_credentials_msg.format(
-			proto_name = self.proto.name,
-			cf_name = (self.proto.is_fork_of or self.proto.name).lower(),
-		))
-
 	# Call family of methods - direct-to-daemon RPC call:
 	# positional params are passed to the daemon, 'timeout' and 'wallet' kwargs to the backend
 

test/unit_tests_d/ut_rpc.py

@@ -3,6 +3,8 @@
 test.unit_tests_d.ut_rpc: RPC unit test for the MMGen suite
 """
 
+import time
+
 from mmgen.common import *
 
 from mmgen.protocol import init_proto
@@ -11,23 +13,32 @@ from mmgen.daemon import CoinDaemon
 from mmgen.proto.xmr.rpc import MoneroRPCClient,MoneroRPCClientRaw,MoneroWalletRPCClient
 from mmgen.proto.xmr.daemon import MoneroWalletDaemon
 
-def cfg_file_auth_test(proto,d):
-	qmsg(cyan(f'\n  Testing authentication with credentials from {d.cfg_file}:'))
+def cfg_file_auth_test(proto,d,bad_auth=False):
+	m = 'missing credentials' if bad_auth else f'credentials from {d.cfg_file}'
+	qmsg(cyan(f'\n  Testing authentication with {m}:'))
+	time.sleep(0.1) # race condition
 	d.remove_datadir() # removes cookie file to force authentication from cfg file
 	os.makedirs(d.network_datadir)
 
-	cf = os.path.join(d.datadir,d.cfg_file)
-	with open(cf,'a') as fp:
-		fp.write('\nrpcuser = ut_rpc\nrpcpassword = ut_rpc_passw0rd\n')
+	if not bad_auth:
+		cf = os.path.join(d.datadir,d.cfg_file)
+		with open(cf,'a') as fp:
+			fp.write('\nrpcuser = ut_rpc\nrpcpassword = ut_rpc_passw0rd\n')
+		d.flag.keep_cfg_file = True
 
-	d.flag.keep_cfg_file = True
 	d.start()
 
-	async def do():
-		rpc = await rpc_init(proto)
+	if bad_auth:
+		os.rename(d.auth_cookie_fn,d.auth_cookie_fn+'.bak')
+		try: async_run(rpc_init(proto))
+		except Exception as e:
+			vmsg(yellow(str(e)))
+		else: die(3,'No error on missing credentials!')
+		os.rename(d.auth_cookie_fn+'.bak',d.auth_cookie_fn)
+	else:
+		rpc = async_run(rpc_init(proto))
 		assert rpc.auth.user == 'ut_rpc', f'{rpc.auth.user}: user is not ut_rpc!'
 
-	async_run(do())
 	d.stop()
 
 def print_daemon_info(rpc):
@@ -105,6 +116,7 @@ def run_test(network_ids,test_cf_auth=False,daemon_ids=None):
 
 		if test_cf_auth and g.platform != 'win':
 			cfg_file_auth_test(d.proto,d)
+			cfg_file_auth_test(d.proto,d,bad_auth=True)
 
 		qmsg('')