fileutil.py 9.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349
  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. fileutil: Routines that read, write, execute or stat files
  20. """
  21. # 09 Mar 2024: make all utils accept Path instances as arguments
  22. import sys,os
  23. from .color import set_vt100
  24. from .util import (
  25. msg,
  26. die,
  27. get_extension,
  28. is_utf8,
  29. capfirst,
  30. make_full_path,
  31. strip_comments,
  32. )
  33. def check_or_create_dir(path):
  34. try:
  35. os.listdir(path)
  36. except:
  37. if os.getenv('MMGEN_TEST_SUITE'):
  38. if os.path.exists(path): # path is a link or regular file
  39. from subprocess import run
  40. run([
  41. ('rm' if sys.platform == 'win32' else '/bin/rm'),
  42. '-rf',
  43. str(path)])
  44. set_vt100()
  45. try:
  46. os.makedirs(path,0o700)
  47. except:
  48. die(2, f'ERROR: unable to read or create path ‘{path}’')
  49. def check_binary(args):
  50. from subprocess import run,DEVNULL
  51. try:
  52. run(args,stdout=DEVNULL,stderr=DEVNULL,check=True)
  53. except:
  54. die(2,f'{args[0]!r} binary missing, not in path, or not executable')
  55. set_vt100()
  56. def shred_file(fn,verbose=False):
  57. check_binary(['shred','--version'])
  58. from subprocess import run
  59. run(
  60. ['shred','--force','--iterations=30','--zero','--remove=wipesync']
  61. + (['--verbose'] if verbose else [])
  62. + [str(fn)],
  63. check=True )
  64. set_vt100()
  65. def _check_file_type_and_access(fname,ftype,blkdev_ok=False):
  66. import stat
  67. access,op_desc = (
  68. (os.W_OK,'writ') if ftype in ('output file','output directory') else
  69. (os.R_OK,'read') )
  70. if ftype == 'output directory':
  71. ok_types = [(stat.S_ISDIR, 'output directory')]
  72. else:
  73. ok_types = [
  74. (stat.S_ISREG,'regular file'),
  75. (stat.S_ISLNK,'symbolic link')
  76. ]
  77. if blkdev_ok:
  78. if not sys.platform in ('win32',):
  79. ok_types.append((stat.S_ISBLK, 'block device'))
  80. try:
  81. mode = os.stat(fname).st_mode
  82. except:
  83. die('FileNotFound', f'Requested {ftype} ‘{fname}’ not found')
  84. for t in ok_types:
  85. if t[0](mode):
  86. break
  87. else:
  88. ok_list = ' or '.join( t[1] for t in ok_types )
  89. die(1, f'Requested {ftype} ‘{fname}’ is not a {ok_list}')
  90. if not os.access(fname,access):
  91. die(1, f'Requested {ftype} ‘{fname}’ is not {op_desc}able by you')
  92. return True
  93. def check_infile(f,blkdev_ok=False):
  94. return _check_file_type_and_access(f,'input file',blkdev_ok=blkdev_ok)
  95. def check_outfile(f,blkdev_ok=False):
  96. return _check_file_type_and_access(f,'output file',blkdev_ok=blkdev_ok)
  97. def check_outdir(f):
  98. return _check_file_type_and_access(f,'output directory')
  99. def get_seed_file(cfg,nargs,wallets=None,invoked_as=None):
  100. wallets = wallets or cfg._args
  101. from .filename import find_file_in_dir
  102. from .wallet.mmgen import wallet
  103. wf = find_file_in_dir(wallet,cfg.data_dir)
  104. wd_from_opt = bool(cfg.hidden_incog_input_params or cfg.in_fmt) # have wallet data from opt?
  105. if len(wallets) + (wd_from_opt or bool(wf)) < nargs:
  106. if not wf:
  107. msg('No default wallet found, and no other seed source was specified')
  108. cfg._opts.usage()
  109. elif len(wallets) > nargs:
  110. cfg._opts.usage()
  111. elif len(wallets) == nargs and wf and invoked_as != 'gen':
  112. cfg._util.qmsg('Warning: overriding default wallet with user-supplied wallet')
  113. if wallets or wf:
  114. check_infile(wallets[0] if wallets else wf)
  115. return str(wallets[0]) if wallets else (wf,None)[wd_from_opt] # could be a Path instance
  116. def _open_or_die(filename,mode,silent=False):
  117. try:
  118. return open(filename,mode)
  119. except:
  120. if silent:
  121. die(2,'')
  122. else:
  123. fn = {0:'STDIN',1:'STDOUT',2:'STDERR'}[filename] if isinstance(filename,int) else f'‘{filename}’'
  124. desc = 'reading' if 'r' in mode else 'writing'
  125. die(2, f'Unable to open file {fn} for {desc}')
  126. def write_data_to_file(
  127. cfg,
  128. outfile,
  129. data,
  130. desc = 'data',
  131. ask_write = False,
  132. ask_write_prompt = '',
  133. ask_write_default_yes = True,
  134. ask_overwrite = True,
  135. ask_tty = True,
  136. no_tty = False,
  137. quiet = False,
  138. binary = False,
  139. ignore_opt_outdir = False,
  140. outdir = None,
  141. check_data = False,
  142. cmp_data = None):
  143. outfile = str(outfile) # could be a Path instance
  144. if quiet:
  145. ask_tty = ask_overwrite = False
  146. if cfg.quiet:
  147. ask_overwrite = False
  148. if ask_write_default_yes is False or ask_write_prompt:
  149. ask_write = True
  150. def do_stdout():
  151. cfg._util.qmsg('Output to STDOUT requested')
  152. if cfg.stdin_tty:
  153. if no_tty:
  154. die(2,f'Printing {desc} to screen is not allowed')
  155. if (ask_tty and not cfg.quiet) or binary:
  156. from .ui import confirm_or_raise
  157. confirm_or_raise(
  158. cfg,
  159. message = '',
  160. action = f'output {desc} to screen' )
  161. else:
  162. try:
  163. of = os.readlink(f'/proc/{os.getpid()}/fd/1') # Linux
  164. except:
  165. of = None # Windows
  166. if of:
  167. if of[:5] == 'pipe:':
  168. if no_tty:
  169. die(2,f'Writing {desc} to pipe is not allowed')
  170. if ask_tty and not cfg.quiet:
  171. from .ui import confirm_or_raise
  172. confirm_or_raise(
  173. cfg,
  174. message = '',
  175. action = f'output {desc} to pipe' )
  176. msg('')
  177. of2,pd = os.path.relpath(of),os.path.pardir
  178. msg('Redirecting output to file {!r}'.format(of if of2[:len(pd)] == pd else of2))
  179. else:
  180. msg('Redirecting output to file')
  181. if binary:
  182. if sys.platform == 'win32': # condition on separate line for pylint
  183. import msvcrt
  184. msvcrt.setmode(sys.stdout.fileno(),os.O_BINARY)
  185. # MSWin workaround. See msg_r()
  186. try:
  187. sys.stdout.write(data.decode() if isinstance(data,bytes) else data)
  188. except:
  189. os.write(1,data if isinstance(data,bytes) else data.encode())
  190. def do_file(outfile,ask_write_prompt):
  191. if (outdir or (cfg.outdir and not ignore_opt_outdir)) and not os.path.isabs(outfile):
  192. outfile = make_full_path(str(outdir or cfg.outdir), outfile) # outdir could be Path instance
  193. if ask_write:
  194. if not ask_write_prompt:
  195. ask_write_prompt = f'Save {desc}?'
  196. from .ui import keypress_confirm
  197. if not keypress_confirm(
  198. cfg,
  199. ask_write_prompt,
  200. default_yes = ask_write_default_yes ):
  201. die(1,f'{capfirst(desc)} not saved')
  202. hush = False
  203. if os.path.lexists(outfile) and ask_overwrite:
  204. from .ui import confirm_or_raise
  205. confirm_or_raise(
  206. cfg,
  207. message = '',
  208. action = f'File {outfile!r} already exists\nOverwrite?' )
  209. msg(f'Overwriting file {outfile!r}')
  210. hush = True
  211. # not atomic, but better than nothing
  212. # if cmp_data is empty, file can be either empty or non-existent
  213. if check_data:
  214. d = ''
  215. try:
  216. with open(outfile,('r','rb')[bool(binary)]) as fp:
  217. d = fp.read()
  218. except:
  219. pass
  220. if d != cmp_data:
  221. die(3,f'{desc} in file {outfile!r} has been altered by some other program! Aborting file write')
  222. # To maintain portability, always open files in binary mode
  223. # If 'binary' option not set, encode/decode data before writing and after reading
  224. try:
  225. with _open_or_die(outfile,'wb') as fp:
  226. fp.write(data if binary else data.encode())
  227. except:
  228. die(2,f'Failed to write {desc} to file {outfile!r}')
  229. if not (hush or quiet):
  230. msg(f'{capfirst(desc)} written to file {outfile!r}')
  231. return True
  232. if cfg.stdout or outfile in ('','-'):
  233. do_stdout()
  234. elif sys.stdin.isatty() and not sys.stdout.isatty():
  235. do_stdout()
  236. else:
  237. do_file(outfile,ask_write_prompt)
  238. def get_words_from_file(cfg,infile,desc,quiet=False):
  239. if not quiet:
  240. cfg._util.qmsg(f'Getting {desc} from file ‘{infile}’')
  241. with _open_or_die(infile, 'rb') as fp:
  242. data = fp.read()
  243. try:
  244. words = data.decode().split()
  245. except:
  246. die(1,f'{capfirst(desc)} data must be UTF-8 encoded.')
  247. cfg._util.dmsg('Sanitized input: [{}]'.format(' '.join(words)))
  248. return words
  249. def get_data_from_file(
  250. cfg,
  251. infile,
  252. desc = 'data',
  253. dash = False,
  254. silent = False,
  255. binary = False,
  256. quiet = False ):
  257. if not (cfg.quiet or silent or quiet):
  258. cfg._util.qmsg(f'Getting {desc} from file ‘{infile}’')
  259. with _open_or_die(
  260. (0 if dash and infile == '-' else infile),
  261. 'rb',
  262. silent=silent) as fp:
  263. data = fp.read(cfg.max_input_size+1)
  264. if not binary:
  265. data = data.decode()
  266. if len(data) == cfg.max_input_size + 1:
  267. die( 'MaxInputSizeExceeded',
  268. f'Too much input data! Max input data size: {cfg.max_input_size} bytes' )
  269. return data
  270. def get_lines_from_file(
  271. cfg,
  272. fn,
  273. desc = 'data',
  274. trim_comments = False,
  275. quiet = False,
  276. silent = False ):
  277. def decrypt_file_maybe():
  278. data = get_data_from_file( cfg, fn, desc=desc, binary=True, quiet=quiet, silent=silent )
  279. from .crypto import Crypto
  280. have_enc_ext = get_extension(fn) == Crypto.mmenc_ext
  281. if have_enc_ext or not is_utf8(data):
  282. m = ('Attempting to decrypt','Decrypting')[have_enc_ext]
  283. cfg._util.qmsg(f'{m} {desc} ‘{fn}’')
  284. data = Crypto(cfg).mmgen_decrypt_retry(data,desc)
  285. return data
  286. lines = decrypt_file_maybe().decode().splitlines()
  287. if trim_comments:
  288. lines = strip_comments(lines)
  289. cfg._util.dmsg(f'Got {len(lines)} lines from file ‘{fn}’')
  290. return lines