ct_autosign.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969
  1. #!/usr/bin/env python3
  2. #
  3. # mmgen = Multi-Mode GENerator, command-line Bitcoin cold storage solution
  4. # Copyright (C)2013-2024 The MMGen Project <mmgen@tuta.io>
  5. #
  6. # This program is free software: you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License as published by
  8. # the Free Software Foundation, either version 3 of the License, or
  9. # (at your option) any later version.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License
  17. # along with this program. If not, see <http://www.gnu.org/licenses/>.
  18. """
  19. test.cmdtest_py_d.ct_autosign: Autosign tests for the cmdtest.py test suite
  20. """
  21. import sys, os, time, shutil, atexit
  22. from subprocess import run,DEVNULL
  23. from pathlib import Path
  24. from mmgen.cfg import Config
  25. from mmgen.color import red,green,blue,yellow,cyan,orange,purple,gray
  26. from mmgen.util import msg,suf,die,indent,fmt
  27. from mmgen.led import LEDControl
  28. from mmgen.autosign import Autosign, Signable
  29. from ..include.common import (
  30. cfg,
  31. omsg,
  32. omsg_r,
  33. oqmsg,
  34. oqmsg_r,
  35. start_test_daemons,
  36. stop_test_daemons,
  37. joinpath,
  38. imsg,
  39. read_from_file,
  40. silence,
  41. end_silence,
  42. VirtBlockDevice,
  43. )
  44. from .common import ref_dir,dfl_words_file,dfl_bip39_file
  45. from .ct_base import CmdTestBase
  46. from .input import stealth_mnemonic_entry
  47. class CmdTestAutosignBase(CmdTestBase):
  48. networks = ('btc',)
  49. tmpdir_nums = [18]
  50. color = True
  51. platform_skip = ('win32',)
  52. threaded = False
  53. daemon_coins = []
  54. def __init__(self,trunner,cfgs,spawn):
  55. CmdTestBase.__init__(self,trunner,cfgs,spawn)
  56. if trunner is None:
  57. return
  58. self.silent_mount = self.live or not (cfg.exact_output or cfg.verbose)
  59. self.network_ids = [c+'_tn' for c in self.daemon_coins] + self.daemon_coins
  60. self._create_autosign_instances(create_dirs=not cfg.skipping_deps)
  61. if sys.platform == 'linux':
  62. self.txdev = VirtBlockDevice(self.asi.fs_image_path, '10M')
  63. if not (cfg.skipping_deps or self.live):
  64. self._create_removable_device()
  65. if sys.platform == 'darwin' and not cfg.no_daemon_stop:
  66. atexit.register(self._macOS_eject_disk, self.asi.dev_label)
  67. self.opts = ['--coins='+','.join(self.coins)]
  68. if not self.live:
  69. self.spawn_env['MMGEN_TEST_SUITE_ROOT_PFX'] = self.tmpdir
  70. if self.threaded:
  71. self.spawn_env['MMGEN_TEST_SUITE_AUTOSIGN_THREADED'] = '1'
  72. def __del__(self):
  73. if hasattr(self,'have_dummy_control_files'):
  74. LEDControl.delete_dummy_control_files()
  75. if hasattr(self, 'txdev'):
  76. del self.txdev
  77. if not cfg.no_daemon_stop:
  78. if sys.platform == 'darwin':
  79. for label in (self.asi.dev_label, self.asi.ramdisk.label):
  80. self._macOS_eject_disk(label)
  81. def _create_autosign_instances(self,create_dirs):
  82. d = {'offline': {'name':'asi'}}
  83. if self.have_online:
  84. d['online'] = {'name':'asi_online'}
  85. for subdir,data in d.items():
  86. asi = Autosign(
  87. Config({
  88. 'coins': ','.join(self.coins),
  89. 'test_suite': True,
  90. 'test_suite_xmr_autosign': self.name == 'CmdTestXMRAutosign',
  91. 'test_suite_autosign_threaded': self.threaded,
  92. 'test_suite_root_pfx': None if self.live else self.tmpdir,
  93. 'online': subdir == 'online',
  94. }))
  95. if create_dirs and not self.live:
  96. for k in ('mountpoint', 'shm_dir', 'wallet_dir', 'dev_label_dir'):
  97. if subdir == 'online' and k in ('shm_dir', 'wallet_dir'):
  98. continue
  99. if sys.platform == 'darwin' and k != 'mountpoint':
  100. continue
  101. getattr(asi, k).mkdir(parents=True, exist_ok=True)
  102. if sys.platform == 'darwin' and k == 'mountpoint':
  103. asi.mountpoint.rmdir()
  104. continue
  105. setattr(self, data['name'], asi)
  106. def _create_removable_device(self):
  107. if sys.platform == 'linux':
  108. self.txdev.create()
  109. self.txdev.attach(silent=True)
  110. cmd = [
  111. '/sbin/mkfs.ext2',
  112. '-E', 'root_owner={}:{}'.format(os.getuid(), os.getgid()),
  113. '-L', self.asi.dev_label,
  114. str(self.txdev.img_path)]
  115. redir = DEVNULL
  116. run(cmd, stdout=redir, stderr=redir, check=True)
  117. self.txdev.detach(silent=True)
  118. elif sys.platform == 'darwin':
  119. cmd = [
  120. 'hdiutil', 'create', '-size', '10M', '-fs', 'exFAT',
  121. '-volname', self.asi.dev_label,
  122. str(self.asi.fs_image_path)]
  123. redir = None if cfg.exact_output or cfg.verbose else DEVNULL
  124. run(cmd, stdout=redir, check=True)
  125. def _macOS_mount_fs_image(self, loc):
  126. time.sleep(0.2)
  127. run(f'hdiutil attach -mountpoint {loc.mountpoint} {loc.fs_image_path}.dmg'.split(), stdout=DEVNULL, check=True)
  128. def _macOS_eject_disk(self, label):
  129. try:
  130. run(['diskutil' ,'eject', label], stdout=DEVNULL, stderr=DEVNULL)
  131. except:
  132. pass
  133. def start_daemons(self):
  134. self.spawn('',msg_only=True)
  135. start_test_daemons(*self.network_ids)
  136. return 'ok'
  137. def stop_daemons(self):
  138. self.spawn('',msg_only=True)
  139. stop_test_daemons(*self.network_ids)
  140. return 'ok'
  141. def run_setup(
  142. self,
  143. mn_type = None,
  144. mn_file = None,
  145. use_dfl_wallet = False,
  146. seed_len = None,
  147. usr_entry_modes = False,
  148. passwd = 'abc',
  149. expect_args = []):
  150. mn_desc = mn_type or 'default'
  151. mn_type = mn_type or 'mmgen'
  152. if sys.platform == 'darwin' and not cfg.no_daemon_stop:
  153. self._macOS_eject_disk(self.asi.ramdisk.label)
  154. self.insert_device()
  155. t = self.spawn(
  156. 'mmgen-autosign',
  157. self.opts
  158. + ([] if mn_desc == 'default' else [f'--mnemonic-fmt={mn_type}'])
  159. + ([f'--seed-len={seed_len}'] if seed_len else [])
  160. + ['setup'],
  161. no_passthru_opts = True)
  162. if use_dfl_wallet:
  163. t.expect( 'Use default wallet for autosigning? (Y/n): ', 'y' )
  164. t.passphrase('MMGen wallet', passwd)
  165. else:
  166. if use_dfl_wallet is not None: # None => no dfl wallet present
  167. t.expect( 'Use default wallet for autosigning? (Y/n): ', 'n' )
  168. mn_file = mn_file or { 'mmgen': dfl_words_file, 'bip39': dfl_bip39_file }[mn_type]
  169. mn = read_from_file(mn_file).strip().split()
  170. if not seed_len:
  171. t.expect('words: ',{ 12:'1', 18:'2', 24:'3' }[len(mn)])
  172. t.expect('OK? (Y/n): ','\n')
  173. from mmgen.mn_entry import mn_entry
  174. entry_mode = 'full'
  175. mne = mn_entry(cfg, mn_type, entry_mode)
  176. if usr_entry_modes:
  177. t.expect('user-configured')
  178. else:
  179. t.expect(
  180. 'Type a number.*: ',
  181. str(mne.entry_modes.index(entry_mode) + 1),
  182. regex = True)
  183. stealth_mnemonic_entry(t,mne,mn,entry_mode)
  184. t.written_to_file('Autosign wallet')
  185. if expect_args:
  186. t.expect(*expect_args)
  187. t.read()
  188. self.remove_device()
  189. if sys.platform == 'darwin' and not cfg.no_daemon_stop:
  190. atexit.register(self._macOS_eject_disk, self.asi.ramdisk.label)
  191. return t
  192. def insert_device(self, asi='asi'):
  193. if self.live:
  194. return
  195. loc = getattr(self, asi)
  196. if sys.platform == 'linux':
  197. loc.dev_label_path.touch()
  198. # self.txdev.attach() # WIP
  199. elif sys.platform == 'darwin':
  200. self._macOS_mount_fs_image(loc)
  201. def remove_device(self, asi='asi'):
  202. if self.live:
  203. return
  204. loc = getattr(self, asi)
  205. if sys.platform == 'linux':
  206. if loc.dev_label_path.exists():
  207. loc.dev_label_path.unlink()
  208. # self.txdev.detach() # WIP
  209. elif sys.platform == 'darwin':
  210. self._macOS_eject_disk(loc.dev_label)
  211. def _mount_ops(self, loc, cmd, *args, **kwargs):
  212. return getattr(getattr(self,loc),cmd)(*args, silent=self.silent_mount, **kwargs)
  213. def do_mount(self, *args, **kwargs):
  214. return self._mount_ops('asi', 'do_mount', *args, **kwargs)
  215. def do_umount(self, *args, **kwargs):
  216. return self._mount_ops('asi', 'do_umount', *args, **kwargs)
  217. def _gen_listing(self):
  218. for k in self.asi.dirs:
  219. d = getattr(self.asi,k)
  220. if d.is_dir():
  221. yield '{:12} {}'.format(
  222. str(Path(*d.parts[6:])) + ':',
  223. ' '.join(sorted(i.name for i in d.iterdir()))).strip()
  224. class CmdTestAutosignClean(CmdTestAutosignBase):
  225. have_online = False
  226. live = False
  227. simulate_led = True
  228. no_insert_check = False
  229. coins = ['btc']
  230. tmpdir_nums = [38]
  231. cmd_group = (
  232. ('clean_no_xmr', 'cleaning signable file directories (no XMR)'),
  233. ('clean_xmr_only', 'cleaning signable file directories (XMR-only)'),
  234. ('clean_all', 'cleaning signable file directories (with XMR)'),
  235. )
  236. def create_fake_tx_files(self):
  237. imsg('Creating fake transaction files')
  238. if not self.asi.xmr_only:
  239. for fn in (
  240. 'a.rawtx', 'a.sigtx',
  241. 'b.rawtx', 'b.sigtx',
  242. 'c.rawtx',
  243. 'd.sigtx',
  244. ):
  245. (self.asi.tx_dir / fn).touch()
  246. for fn in (
  247. 'a.arawtx', 'a.asigtx', 'a.asubtx',
  248. 'b.arawtx', 'b.asigtx',
  249. 'c.asubtx',
  250. 'd.arawtx', 'd.asubtx',
  251. 'e.arawtx',
  252. 'f.asigtx', 'f.asubtx',
  253. ):
  254. (self.asi.txauto_dir / fn).touch()
  255. for fn in (
  256. 'a.rawmsg.json', 'a.sigmsg.json',
  257. 'b.rawmsg.json',
  258. 'c.sigmsg.json',
  259. 'd.rawmsg.json', 'd.sigmsg.json',
  260. ):
  261. (self.asi.msg_dir / fn).touch()
  262. if self.asi.have_xmr:
  263. for fn in (
  264. 'a.rawtx', 'a.sigtx', 'a.subtx',
  265. 'b.rawtx', 'b.sigtx',
  266. 'c.subtx',
  267. 'd.rawtx', 'd.subtx',
  268. 'e.rawtx',
  269. 'f.sigtx', 'f.subtx',
  270. ):
  271. (self.asi.xmr_tx_dir / fn).touch()
  272. for fn in (
  273. 'a.raw', 'a.sig',
  274. 'b.raw',
  275. 'c.sig',
  276. ):
  277. (self.asi.xmr_outputs_dir / fn).touch()
  278. return 'ok'
  279. def clean_no_xmr(self):
  280. return self._clean('btc,ltc,eth')
  281. def clean_xmr_only(self):
  282. self.asi = Autosign(Config({'_clone': self.asi.cfg, 'coins': 'xmr'}))
  283. return self._clean('xmr')
  284. def clean_all(self):
  285. self.asi = Autosign(Config({'_clone': self.asi.cfg, 'coins': 'xmr,btc,bch,eth'}))
  286. return self._clean('xmr,btc,bch,eth')
  287. def _clean(self,coins):
  288. self.spawn('', msg_only=True)
  289. self.insert_device()
  290. silence()
  291. self.do_mount()
  292. end_silence()
  293. self.create_fake_tx_files()
  294. before = '\n'.join(self._gen_listing())
  295. t = self.spawn('mmgen-autosign', [f'--coins={coins}','clean'], no_msg=True)
  296. out = t.read()
  297. self.insert_device()
  298. silence()
  299. self.do_mount()
  300. end_silence()
  301. after = '\n'.join(self._gen_listing())
  302. chk_non_xmr = """
  303. tx: a.sigtx b.sigtx c.rawtx d.sigtx
  304. txauto: a.asubtx b.asigtx c.asubtx d.asubtx e.arawtx f.asubtx
  305. msg: a.sigmsg.json b.rawmsg.json c.sigmsg.json d.sigmsg.json
  306. """
  307. chk_xmr = """
  308. xmr: outputs tx
  309. xmr/tx: a.subtx b.sigtx c.subtx d.subtx e.rawtx f.subtx
  310. xmr/outputs:
  311. """
  312. chk = ''
  313. shred_count = 0
  314. if not self.asi.xmr_only:
  315. for k in ('tx_dir', 'txauto_dir', 'msg_dir'):
  316. shutil.rmtree(getattr(self.asi, k))
  317. chk += chk_non_xmr.rstrip()
  318. shred_count += 9
  319. if self.asi.have_xmr:
  320. shutil.rmtree(self.asi.xmr_dir)
  321. chk += chk_xmr.rstrip()
  322. shred_count += 9
  323. self.do_umount()
  324. self.remove_device()
  325. imsg(f'\nBefore cleaning:\n{before}')
  326. imsg(f'\nAfter cleaning:\n{after}')
  327. assert f'{shred_count} files shredded' in out
  328. assert after + '\n' == fmt(chk), f'\n{after}\n!=\n{fmt(chk)}'
  329. return t
  330. class CmdTestAutosignThreaded(CmdTestAutosignBase):
  331. have_online = True
  332. live = False
  333. no_insert_check = False
  334. threaded = True
  335. def _wait_loop_start(self):
  336. t = self.spawn(
  337. 'mmgen-autosign',
  338. self.opts + ['--full-summary', 'wait'],
  339. direct_exec = True,
  340. no_passthru_opts = True,
  341. spawn_env_override = self.spawn_env | {'EXEC_WRAPPER_DO_RUNTIME_MSG': ''})
  342. self.write_to_tmpfile('autosign_thread_pid', str(t.ep.pid))
  343. return t
  344. def wait_loop_start(self):
  345. import threading
  346. threading.Thread(target=self._wait_loop_start, name='Autosign wait loop').start()
  347. time.sleep(0.1) # try to ensure test output is displayed before next test starts
  348. return 'silent'
  349. def wait_loop_kill(self):
  350. self.spawn('',msg_only=True)
  351. pid = int(self.read_from_tmpfile('autosign_thread_pid'))
  352. self.delete_tmpfile('autosign_thread_pid')
  353. from signal import SIGTERM
  354. imsg(purple(f'Killing autosign wait loop [PID {pid}]'))
  355. try:
  356. os.kill(pid,SIGTERM)
  357. except:
  358. imsg(yellow(f'{pid}: no such process'))
  359. return 'ok'
  360. def _wait_signed(self,desc):
  361. oqmsg_r(gray(f'→ offline wallet{"s" if desc.endswith("s") else ""} waiting for {desc}'))
  362. assert not self.asi.device_inserted, (
  363. f'‘{getattr(self.asi, "dev_label_path", self.asi.dev_label)}’ is inserted!')
  364. assert not self.asi.mountpoint.is_mount(), f'‘{self.asi.mountpoint}’ is mounted!'
  365. self.insert_device()
  366. while True:
  367. oqmsg_r(gray('.'))
  368. if self.asi.mountpoint.is_mount():
  369. oqmsg_r(gray(' signing '))
  370. break
  371. time.sleep(0.2)
  372. while True:
  373. oqmsg_r(gray('>'))
  374. if not self.asi.mountpoint.is_mount():
  375. oqmsg(gray(' done'))
  376. break
  377. time.sleep(0.2)
  378. imsg('')
  379. self.remove_device()
  380. return 'ok'
  381. def insert_device_online(self):
  382. return self.insert_device(asi='asi_online')
  383. def remove_device_online(self):
  384. return self.remove_device(asi='asi_online')
  385. def do_mount_online(self, *args, **kwargs):
  386. return self._mount_ops('asi_online', 'do_mount', *args, **kwargs)
  387. def do_umount_online(self, *args, **kwargs):
  388. return self._mount_ops('asi_online', 'do_umount', *args, **kwargs)
  389. async def txview(self):
  390. self.spawn('', msg_only=True)
  391. self.insert_device()
  392. self.do_mount()
  393. src = Path(self.asi.txauto_dir)
  394. from mmgen.tx import CompletedTX
  395. txs = sorted(
  396. [await CompletedTX(cfg=cfg, filename=path, quiet_open=True) for path in sorted(src.iterdir())],
  397. key = lambda x: x.timestamp)
  398. for tx in txs:
  399. imsg(blue(f'\nViewing ‘{tx.infile.name}’:'))
  400. out = tx.info.format(terse=True)
  401. imsg(indent(out, indent=' '))
  402. self.do_umount()
  403. self.remove_device()
  404. return 'ok'
  405. class CmdTestAutosign(CmdTestAutosignBase):
  406. 'autosigning transactions for all supported coins'
  407. coins = ['btc','bch','ltc','eth']
  408. daemon_coins = ['btc','bch','ltc']
  409. txfile_coins = ['btc','bch','ltc','eth','mm1','etc']
  410. have_online = False
  411. live = False
  412. simulate_led = True
  413. no_insert_check = True
  414. filedir_map = (
  415. ('btc',''),
  416. ('bch',''),
  417. ('ltc','litecoin'),
  418. ('eth','ethereum'),
  419. ('mm1','ethereum'),
  420. ('etc','ethereum_classic'),
  421. )
  422. cmd_group = (
  423. ('start_daemons', 'starting daemons'),
  424. ('copy_tx_files', 'copying transaction files'),
  425. ('gen_key', 'generating key'),
  426. ('create_dfl_wallet', 'creating default MMGen wallet'),
  427. ('bad_opt1', 'running ‘mmgen-autosign’ with --seed-len in invalid context'),
  428. ('bad_opt2', 'running ‘mmgen-autosign’ with --mnemonic-fmt in invalid context'),
  429. ('bad_opt3', 'running ‘mmgen-autosign’ with --led in invalid context'),
  430. ('run_setup_dfl_wallet', 'running ‘autosign setup’ (with default wallet)'),
  431. ('sign_quiet', 'signing transactions (--quiet)'),
  432. ('remove_signed_txfiles', 'removing signed transaction files'),
  433. ('run_setup_bip39', 'running ‘autosign setup’ (BIP39 mnemonic)'),
  434. ('create_bad_txfiles', 'creating bad transaction files'),
  435. ('sign_full_summary', 'signing transactions (--full-summary)'),
  436. ('remove_signed_txfiles_btc','removing transaction files (BTC only)'),
  437. ('remove_bad_txfiles', 'removing bad transaction files'),
  438. ('sign_led', 'signing transactions (--led - BTC files only)'),
  439. ('remove_signed_txfiles', 'removing signed transaction files'),
  440. ('sign_stealth_led', 'signing transactions (--stealth-led)'),
  441. ('remove_signed_txfiles', 'removing signed transaction files'),
  442. ('copy_msgfiles', 'copying message files'),
  443. ('sign_quiet_msg', 'signing transactions and messages (--quiet)'),
  444. ('remove_signed_txfiles', 'removing signed transaction files'),
  445. ('create_bad_txfiles2', 'creating bad transaction files'),
  446. ('remove_signed_msgfiles', 'removing signed message files'),
  447. ('create_invalid_msgfile', 'creating invalid message file'),
  448. ('sign_full_summary_msg', 'signing transactions and messages (--full-summary)'),
  449. ('remove_invalid_msgfile', 'removing invalid message file'),
  450. ('remove_bad_txfiles2', 'removing bad transaction files'),
  451. ('sign_no_unsigned', 'signing transactions and messages (nothing to sign)'),
  452. ('sign_no_unsigned_xmr', 'signing transactions and messages (nothing to sign, with XMR)'),
  453. ('sign_no_unsigned_xmronly', 'signing transactions and messages (nothing to sign, XMR-only)'),
  454. ('wipe_key', 'wiping the wallet encryption key'),
  455. ('stop_daemons', 'stopping daemons'),
  456. ('sign_bad_no_daemon', 'signing transactions (error, no daemons running)'),
  457. )
  458. def __init__(self,trunner,cfgs,spawn):
  459. super().__init__(trunner,cfgs,spawn)
  460. if trunner is None:
  461. return
  462. if self.live and not cfg.exact_output:
  463. die(1,red('autosign_live tests must be run with --exact-output enabled!'))
  464. if self.no_insert_check:
  465. self.opts.append('--no-insert-check')
  466. self.tx_file_ops('set_count') # initialize self.tx_count here so we can resume anywhere
  467. self.bad_tx_count = 0
  468. def gen_msg_fns():
  469. fmap = dict(self.filedir_map)
  470. for coin in self.coins:
  471. if coin == 'xmr':
  472. continue
  473. sdir = os.path.join('test','ref',fmap[coin])
  474. for fn in os.listdir(sdir):
  475. if fn.endswith(f'[{coin.upper()}].rawmsg.json'):
  476. yield os.path.join(sdir,fn)
  477. self.ref_msgfiles = tuple(gen_msg_fns())
  478. self.good_msg_count = 0
  479. self.bad_msg_count = 0
  480. if self.simulate_led:
  481. LEDControl.create_dummy_control_files()
  482. self.have_dummy_control_files = True
  483. self.spawn_env['MMGEN_TEST_SUITE_AUTOSIGN_LED_SIMULATE'] = '1'
  484. def gen_key(self):
  485. self.insert_device()
  486. t = self.spawn( 'mmgen-autosign', self.opts + ['gen_key'] )
  487. t.expect_getend('Wrote key file ')
  488. t.read()
  489. self.remove_device()
  490. return t
  491. def create_dfl_wallet(self):
  492. t = self.spawn( 'mmgen-walletconv', [
  493. f'--outdir={cfg.data_dir}',
  494. '--usr-randchars=0', '--quiet', '--hash-preset=1', '--label=foo',
  495. 'test/ref/98831F3A.hex'
  496. ]
  497. )
  498. t.passphrase_new('new MMGen wallet','abc')
  499. t.written_to_file('MMGen wallet')
  500. return t
  501. def _bad_opt(self, cmdline, expect):
  502. t = self.spawn('mmgen-autosign', ['--coins=btc'] + cmdline, exit_val=1)
  503. t.expect(expect)
  504. return t
  505. def bad_opt1(self):
  506. return self._bad_opt(['--seed-len=128'], 'makes sense')
  507. def bad_opt2(self):
  508. return self._bad_opt(['--mnemonic-fmt=bip39', 'wait'], 'makes sense')
  509. def bad_opt3(self):
  510. return self._bad_opt(['--led', 'gen_key'], 'makes no sense')
  511. def run_setup_dfl_wallet(self):
  512. return self.run_setup(mn_type='default',use_dfl_wallet=True)
  513. def run_setup_bip39(self):
  514. from mmgen.cfgfile import mmgen_cfg_file
  515. fn = mmgen_cfg_file(cfg,'usr').fn
  516. old_data = mmgen_cfg_file(cfg,'usr').get_data(fn)
  517. new_data = [d.replace('bip39:fixed','bip39:full')[2:]
  518. if d.startswith('# mnemonic_entry_modes') else d for d in old_data]
  519. with open(fn, 'w') as fh:
  520. fh.write('\n'.join(new_data) + '\n')
  521. t = self.run_setup(
  522. mn_type = 'bip39',
  523. seed_len = 256,
  524. usr_entry_modes = True)
  525. with open(fn, 'w') as fh:
  526. fh.write('\n'.join(old_data) + '\n')
  527. return t
  528. def copy_tx_files(self):
  529. self.spawn('',msg_only=True)
  530. return self.tx_file_ops('copy')
  531. def remove_signed_txfiles(self):
  532. self.tx_file_ops('remove_signed')
  533. return 'skip'
  534. def remove_signed_txfiles_btc(self):
  535. self.tx_file_ops('remove_signed',txfile_coins=['btc'])
  536. return 'skip'
  537. def tx_file_ops(self,op,txfile_coins=[]):
  538. assert op in ('copy','set_count','remove_signed')
  539. from .ct_ref import CmdTestRef
  540. def gen():
  541. d = CmdTestRef.sources['ref_tx_file']
  542. dirmap = [e for e in self.filedir_map if e[0] in (txfile_coins or self.txfile_coins)]
  543. for coin,coindir in dirmap:
  544. for network in (0,1):
  545. fn = d[coin][network]
  546. if fn:
  547. yield (coindir,fn)
  548. data = list(gen()) + [('','25EFA3[2.34].testnet.rawtx')] # TX with 2 non-MMGen outputs
  549. self.tx_count = len(data)
  550. if op == 'set_count':
  551. return
  552. self.insert_device()
  553. silence()
  554. self.do_mount(verbose=cfg.verbose or cfg.exact_output)
  555. end_silence()
  556. for coindir,fn in data:
  557. src = joinpath(ref_dir,coindir,fn)
  558. if cfg.debug_utf8:
  559. ext = '.testnet.rawtx' if fn.endswith('.testnet.rawtx') else '.rawtx'
  560. fn = fn[:-len(ext)] + '-α' + ext
  561. target = joinpath(self.asi.tx_dir, fn)
  562. if not op == 'remove_signed':
  563. shutil.copyfile(src,target)
  564. try:
  565. os.unlink(target.replace('.rawtx','.sigtx'))
  566. except:
  567. pass
  568. self.do_umount()
  569. self.remove_device()
  570. return 'ok'
  571. def create_bad_txfiles(self):
  572. return self.bad_txfiles('create')
  573. def remove_bad_txfiles(self):
  574. return self.bad_txfiles('remove')
  575. create_bad_txfiles2 = create_bad_txfiles
  576. remove_bad_txfiles2 = remove_bad_txfiles
  577. def bad_txfiles(self,op):
  578. self.insert_device()
  579. self.do_mount()
  580. # create or delete 2 bad tx files
  581. self.spawn('',msg_only=True)
  582. fns = [joinpath(self.asi.tx_dir, f'bad{n}.rawtx') for n in (1,2)]
  583. if op == 'create':
  584. for fn in fns:
  585. with open(fn,'w') as fp:
  586. fp.write('bad tx data\n')
  587. self.bad_tx_count = 2
  588. elif op == 'remove':
  589. for fn in fns:
  590. try:
  591. os.unlink(fn)
  592. except:
  593. pass
  594. self.bad_tx_count = 0
  595. self.do_umount()
  596. self.remove_device()
  597. return 'ok'
  598. def copy_msgfiles(self):
  599. return self.msgfile_ops('copy')
  600. def remove_signed_msgfiles(self):
  601. return self.msgfile_ops('remove_signed')
  602. def create_invalid_msgfile(self):
  603. return self.msgfile_ops('create_invalid')
  604. def remove_invalid_msgfile(self):
  605. return self.msgfile_ops('remove_invalid')
  606. def msgfile_ops(self,op):
  607. self.spawn('',msg_only=True)
  608. destdir = joinpath(self.asi.mountpoint,'msg')
  609. self.insert_device()
  610. self.do_mount()
  611. os.makedirs(destdir,exist_ok=True)
  612. if op.endswith('_invalid'):
  613. fn = os.path.join(destdir,'DEADBE[BTC].rawmsg.json')
  614. if op == 'create_invalid':
  615. with open(fn,'w') as fp:
  616. fp.write('bad data\n')
  617. self.bad_msg_count += 1
  618. elif op == 'remove_invalid':
  619. os.unlink(fn)
  620. self.bad_msg_count -= 1
  621. else:
  622. for fn in self.ref_msgfiles:
  623. if op == 'copy':
  624. if os.path.basename(fn) == 'ED405C[BTC].rawmsg.json': # contains bad Seed ID
  625. self.bad_msg_count += 1
  626. else:
  627. self.good_msg_count += 1
  628. imsg(f'Copying: {fn} -> {destdir}')
  629. shutil.copy2(fn,destdir)
  630. elif op == 'remove_signed':
  631. os.unlink(os.path.join( destdir, os.path.basename(fn).replace('rawmsg','sigmsg') ))
  632. self.do_umount()
  633. self.remove_device()
  634. return 'ok'
  635. def do_sign(self, args=[], have_msg=False, exc_exit_val=None):
  636. tx_desc = Signable.transaction.desc
  637. self.insert_device()
  638. t = self.spawn(
  639. 'mmgen-autosign',
  640. self.opts + args,
  641. exit_val = exc_exit_val or (1 if self.bad_tx_count or (have_msg and self.bad_msg_count) else None))
  642. if exc_exit_val:
  643. return t
  644. t.expect(
  645. f'{self.tx_count} {tx_desc}{suf(self.tx_count)} signed' if self.tx_count else
  646. f'No unsigned {tx_desc}s')
  647. if self.bad_tx_count:
  648. t.expect(f'{self.bad_tx_count} {tx_desc}{suf(self.bad_tx_count)} failed to sign')
  649. if have_msg:
  650. t.expect(
  651. f'{self.good_msg_count} message file{suf(self.good_msg_count)}{{0,1}} signed'
  652. if self.good_msg_count else
  653. 'No unsigned message files', regex=True)
  654. if self.bad_msg_count:
  655. t.expect(
  656. f'{self.bad_msg_count} message file{suf(self.bad_msg_count)}{{0,1}} failed to sign',
  657. regex = True)
  658. t.read()
  659. self.remove_device()
  660. imsg('')
  661. return t
  662. def sign_quiet(self):
  663. return self.do_sign(['--quiet'])
  664. def sign_full_summary(self):
  665. return self.do_sign(['--full-summary'])
  666. def sign_led(self):
  667. return self.do_sign(['--quiet', '--led'])
  668. def sign_stealth_led(self):
  669. return self.do_sign(['--quiet', '--stealth-led'])
  670. def sign_quiet_msg(self):
  671. return self.do_sign(['--quiet'], have_msg=True)
  672. def sign_full_summary_msg(self):
  673. return self.do_sign(['--full-summary'], have_msg=True)
  674. def sign_bad_no_daemon(self):
  675. t = self.do_sign(exc_exit_val=2)
  676. t.expect('listening on the correct port')
  677. return t
  678. def sign_no_unsigned(self):
  679. return self._sign_no_unsigned(
  680. coins = 'BTC',
  681. present = ['non_xmr_signables'],
  682. absent = ['xmr_signables'])
  683. def sign_no_unsigned_xmr(self):
  684. if self.coins == ['btc']:
  685. return 'skip'
  686. return self._sign_no_unsigned(
  687. coins = 'XMR,BTC',
  688. present = ['xmr_signables','non_xmr_signables'])
  689. def sign_no_unsigned_xmronly(self):
  690. if self.coins == ['btc']:
  691. return 'skip'
  692. return self._sign_no_unsigned(
  693. coins = 'XMR',
  694. present = ['xmr_signables'],
  695. absent = ['non_xmr_signables'])
  696. def _sign_no_unsigned(self,coins,present=[],absent=[]):
  697. self.insert_device()
  698. t = self.spawn('mmgen-autosign', ['--quiet', '--no-insert-check', f'--coins={coins}'])
  699. res = t.read()
  700. self.remove_device()
  701. for signable_list in present:
  702. for signable_clsname in getattr(Signable,signable_list):
  703. desc = getattr(Signable, signable_clsname).desc
  704. assert f'No unsigned {desc}s' in res, f'‘No unsigned {desc}s’ missing in output'
  705. for signable_list in absent:
  706. for signable_clsname in getattr(Signable,signable_list):
  707. desc = getattr(Signable, signable_clsname).desc
  708. assert not f'No unsigned {desc}s' in res, f'‘No unsigned {desc}s’ should be absent in output'
  709. return t
  710. def wipe_key(self):
  711. self.insert_device()
  712. t = self.spawn('mmgen-autosign', ['--quiet', '--no-insert-check', 'wipe_key'])
  713. t.expect('Shredding')
  714. t.read()
  715. self.remove_device()
  716. return t
  717. class CmdTestAutosignBTC(CmdTestAutosign):
  718. 'autosigning BTC transactions'
  719. coins = ['btc']
  720. daemon_coins = ['btc']
  721. txfile_coins = ['btc']
  722. class CmdTestAutosignLive(CmdTestAutosignBTC):
  723. 'live autosigning BTC transactions'
  724. live = True
  725. simulate_led = False
  726. no_insert_check = False
  727. cmd_group = (
  728. ('start_daemons', 'starting daemons'),
  729. ('copy_tx_files', 'copying transaction files'),
  730. ('gen_key', 'generating key'),
  731. ('run_setup_mmgen', 'running ‘autosign setup’ (MMGen native mnemonic)'),
  732. ('sign_live', 'signing transactions'),
  733. ('create_bad_txfiles', 'creating bad transaction files'),
  734. ('sign_live_led', 'signing transactions (--led)'),
  735. ('remove_bad_txfiles', 'removing bad transaction files'),
  736. ('sign_live_stealth_led','signing transactions (--stealth-led)'),
  737. ('stop_daemons', 'stopping daemons'),
  738. )
  739. def __init__(self,trunner,cfgs,spawn):
  740. super().__init__(trunner,cfgs,spawn)
  741. if trunner is None:
  742. return
  743. try:
  744. cf = LEDControl(enabled=True,simulate=self.simulate_led)
  745. except Exception as e:
  746. msg(str(e))
  747. die(2,'LEDControl initialization failed')
  748. for path in (cf.board.status,cf.board.trigger):
  749. if path:
  750. run(['sudo','chmod','0666',path],check=True)
  751. def run_setup_mmgen(self):
  752. return self.run_setup(mn_type='mmgen',use_dfl_wallet=None)
  753. def sign_live(self):
  754. return self.do_sign_live()
  755. def sign_live_led(self):
  756. return self.do_sign_live(['--led'], 'The LED should start blinking slowly now')
  757. def sign_live_stealth_led(self):
  758. return self.do_sign_live(['--stealth-led'], 'You should see no LED activity now')
  759. def do_sign_live(self,led_opts=None,led_msg=None):
  760. def prompt_remove():
  761. omsg_r(orange('\nExtract removable device and then hit ENTER '))
  762. input()
  763. def prompt_insert_sign(t):
  764. omsg(orange(insert_msg))
  765. t.expect(f'{self.tx_count} non-automount transactions signed')
  766. if self.bad_tx_count:
  767. t.expect(f'{self.bad_tx_count} non-automount transactions failed to sign')
  768. t.expect('Waiting')
  769. if led_opts:
  770. opts_msg = '‘' + ' '.join(led_opts) + '’'
  771. info_msg = 'Running ‘mmgen-autosign wait’ with {}. {}'.format(opts_msg, led_msg)
  772. insert_msg = 'Insert removable device and watch for fast LED activity during signing'
  773. else:
  774. opts_msg = 'no LED'
  775. info_msg = 'Running ‘mmgen-autosign wait’'
  776. insert_msg = 'Insert removable device '
  777. self.spawn('', msg_only=True)
  778. self.do_umount()
  779. prompt_remove()
  780. omsg('\n' + cyan(indent(info_msg)))
  781. t = self.spawn(
  782. 'mmgen-autosign',
  783. self.opts + (led_opts or []) + ['--quiet', '--no-summary', 'wait'],
  784. no_msg = True,
  785. exit_val = 1)
  786. if not cfg.exact_output:
  787. omsg('')
  788. prompt_insert_sign(t)
  789. self.do_mount() # race condition due to device insertion detection
  790. self.remove_signed_txfiles()
  791. self.do_umount()
  792. imsg(purple('\nKilling wait loop!'))
  793. t.kill(2) # 2 = SIGINT
  794. if self.simulate_led and led_opts:
  795. t.expect('Stopping LED')
  796. return t
  797. class CmdTestAutosignLiveSimulate(CmdTestAutosignLive):
  798. 'live autosigning BTC transactions with simulated LED support'
  799. simulate_led = True