file.py 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317
  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. tx.file: Transaction file operations for the MMGen suite
  20. """
  21. import os, json
  22. from ..util import ymsg,make_chksum_6,die
  23. from ..obj import MMGenObject,HexStr,MMGenTxID,CoinTxID,MMGenTxComment
  24. from ..rpc import json_encoder
  25. def json_dumps(data):
  26. return json.dumps(data, separators = (',', ':'), cls=json_encoder)
  27. def get_proto_from_coin_id(tx, coin_id, chain):
  28. coin, tokensym = coin_id.split(':') if ':' in coin_id else (coin_id, None)
  29. from ..protocol import CoinProtocol, init_proto
  30. network = CoinProtocol.Base.chain_name_to_network(tx.cfg, coin, chain)
  31. proto = init_proto(tx.cfg, coin, network=network, need_amt=True)
  32. if tokensym:
  33. proto.tokensym = tokensym
  34. return proto
  35. def eval_io_data(tx, data, desc):
  36. if not (desc == 'outputs' and tx.proto.base_coin == 'ETH'): # ETH txs can have no outputs
  37. assert len(data), f'no {desc}!'
  38. for d in data:
  39. d['amt'] = tx.proto.coin_amt(d['amt'])
  40. io, io_list = {
  41. 'inputs': (tx.Input, tx.InputList),
  42. 'outputs': (tx.Output, tx.OutputList),
  43. }[desc]
  44. return io_list(parent=tx, data=[io(tx.proto,**d) for d in data])
  45. class MMGenTxFile(MMGenObject):
  46. data_label = 'MMGenTransaction'
  47. attrs = {
  48. 'chain': None,
  49. 'txid': MMGenTxID,
  50. 'send_amt': 'skip',
  51. 'timestamp': None,
  52. 'blockcount': None,
  53. 'serialized': None,
  54. }
  55. extra_attrs = {
  56. 'locktime': None,
  57. 'comment': MMGenTxComment,
  58. 'coin_txid': CoinTxID,
  59. 'sent_timestamp': None,
  60. }
  61. def __init__(self,tx):
  62. self.tx = tx
  63. self.fmt_data = None
  64. self.filename = None
  65. def parse(self, infile, metadata_only=False, quiet_open=False):
  66. tx = self.tx
  67. from ..fileutil import get_data_from_file
  68. data = get_data_from_file(tx.cfg, infile, f'{tx.desc} data', quiet=quiet_open)
  69. if len(data) > tx.cfg.max_tx_file_size:
  70. die('MaxFileSizeExceeded',
  71. f'Transaction file size exceeds limit ({tx.cfg.max_tx_file_size} bytes)')
  72. return (self.parse_data_json if data[0] == '{' else self.parse_data_legacy)(data, metadata_only)
  73. def parse_data_json(self, data, metadata_only):
  74. tx = self.tx
  75. tx.file_format = 'json'
  76. outer_data = json.loads(data)
  77. data = outer_data[self.data_label]
  78. if outer_data['chksum'] != make_chksum_6(json_dumps(data)):
  79. chk = make_chksum_6(json_dumps(data))
  80. die(3, f'{self.data_label}: invalid checksum for TxID {data["txid"]} ({chk} != {outer_data["chksum"]})')
  81. tx.proto = get_proto_from_coin_id(tx, data['coin_id'], data['chain'])
  82. for k, v in self.attrs.items():
  83. if v != 'skip':
  84. setattr(tx, k, v(data[k]) if v else data[k])
  85. if metadata_only:
  86. return
  87. for k, v in self.extra_attrs.items():
  88. if k in data:
  89. setattr(tx, k, v(data[k]) if v else data[k])
  90. for k in ('inputs', 'outputs'):
  91. setattr(tx, k, eval_io_data(tx, data[k], k))
  92. tx.check_txfile_hex_data()
  93. tx.parse_txfile_serialized_data() # Ethereum RLP or JSON data
  94. assert tx.proto.coin_amt(data['send_amt']) == tx.send_amt, f'{data["send_amt"]} != {tx.send_amt}'
  95. def parse_data_legacy(self, data, metadata_only):
  96. tx = self.tx
  97. tx.file_format = 'legacy'
  98. def deserialize(raw_data, desc):
  99. from ast import literal_eval
  100. try:
  101. return literal_eval(raw_data)
  102. except:
  103. if desc == 'inputs':
  104. ymsg('Warning: transaction data appears to be in old format')
  105. import re
  106. return literal_eval(re.sub(r"[A-Za-z]+?\(('.+?')\)",r'\1', raw_data))
  107. desc = 'data'
  108. try:
  109. tx_data = data.splitlines()
  110. assert len(tx_data) >= 5,'number of lines less than 5'
  111. assert len(tx_data[0]) == 6,'invalid length of first line'
  112. assert HexStr(tx_data.pop(0)) == make_chksum_6(' '.join(tx_data)), 'file data does not match checksum'
  113. if len(tx_data) == 7:
  114. desc = 'sent timestamp'
  115. (_, tx.sent_timestamp) = tx_data.pop(-1).split()
  116. assert _ == 'Sent', 'invalid sent timestamp line'
  117. if len(tx_data) == 6:
  118. assert len(tx_data[-1]) == 64,'invalid coin TxID length'
  119. desc = 'coin TxID'
  120. tx.coin_txid = CoinTxID(tx_data.pop(-1))
  121. if len(tx_data) == 5:
  122. # rough check: allow for 4-byte utf8 characters + base58 (4 * 11 / 8 = 6 (rounded up))
  123. assert len(tx_data[-1]) < MMGenTxComment.max_len*6,'invalid comment length'
  124. c = tx_data.pop(-1)
  125. if c != '-':
  126. desc = 'encoded comment (not base58)'
  127. from ..baseconv import baseconv
  128. comment = baseconv('b58').tobytes(c).decode()
  129. assert comment is not False,'invalid comment'
  130. desc = 'comment'
  131. tx.comment = MMGenTxComment(comment)
  132. desc = 'number of lines' # four required lines
  133. io_data = {}
  134. (metadata, tx.serialized, io_data['inputs'], io_data['outputs']) = tx_data
  135. assert len(metadata) < 100,'invalid metadata length' # rough check
  136. metadata = metadata.split()
  137. if metadata[-1].startswith('LT='):
  138. desc = 'locktime'
  139. tx.locktime = int(metadata.pop()[3:])
  140. desc = 'coin token in metadata'
  141. coin_id = metadata.pop(0) if len(metadata) == 6 else 'BTC'
  142. desc = 'chain token in metadata'
  143. tx.chain = metadata.pop(0).lower() if len(metadata) == 5 else 'mainnet'
  144. desc = 'coin_id or chain'
  145. tx.proto = get_proto_from_coin_id(tx, coin_id, tx.chain)
  146. desc = 'metadata (4 items)'
  147. (txid, send_amt, tx.timestamp, blockcount) = metadata
  148. desc = 'TxID in metadata'
  149. tx.txid = MMGenTxID(txid)
  150. desc = 'block count in metadata'
  151. tx.blockcount = int(blockcount)
  152. if metadata_only:
  153. return
  154. desc = 'transaction file hex data'
  155. tx.check_txfile_hex_data()
  156. desc = 'Ethereum RLP or JSON data'
  157. tx.parse_txfile_serialized_data()
  158. for k in ('inputs', 'outputs'):
  159. desc = f'{k} data'
  160. res = deserialize(io_data[k], k)
  161. for d in res:
  162. if 'label' in d:
  163. d['comment'] = d['label']
  164. del d['label']
  165. setattr(tx, k, eval_io_data(tx, res, k))
  166. desc = 'send amount in metadata'
  167. assert tx.proto.coin_amt(send_amt) == tx.send_amt, f'{send_amt} != {tx.send_amt}'
  168. except Exception as e:
  169. die(2,f'Invalid {desc} in transaction file: {e!s}')
  170. def make_filename(self):
  171. tx = self.tx
  172. def gen_filename():
  173. yield tx.txid
  174. if tx.coin != 'BTC':
  175. yield '-' + tx.dcoin
  176. yield f'[{tx.send_amt!s}'
  177. if tx.is_replaceable():
  178. yield ',{}'.format(tx.fee_abs2rel(tx.fee,to_unit=tx.fn_fee_unit))
  179. if tx.get_serialized_locktime():
  180. yield f',tl={tx.get_serialized_locktime()}'
  181. yield ']'
  182. if tx.proto.testnet:
  183. yield '.' + tx.proto.network
  184. yield '.' + tx.ext
  185. return ''.join(gen_filename())
  186. def format(self):
  187. tx = self.tx
  188. coin_id = tx.coin + ('' if tx.coin == tx.dcoin else ':'+tx.dcoin)
  189. def format_data_legacy():
  190. def amt_to_str(d):
  191. return {k: (str(d[k]) if k == 'amt' else d[k]) for k in d}
  192. lines = [
  193. '{}{} {} {} {} {}{}'.format(
  194. (f'{coin_id} ' if coin_id and tx.coin != 'BTC' else ''),
  195. tx.chain.upper(),
  196. tx.txid,
  197. tx.send_amt,
  198. tx.timestamp,
  199. tx.blockcount,
  200. (f' LT={tx.locktime}' if tx.locktime else ''),
  201. ),
  202. tx.serialized,
  203. ascii([amt_to_str(e._asdict()) for e in tx.inputs]),
  204. ascii([amt_to_str(e._asdict()) for e in tx.outputs])
  205. ]
  206. if tx.comment:
  207. from ..baseconv import baseconv
  208. lines.append(baseconv('b58').frombytes(tx.comment.encode(),tostr=True))
  209. if tx.coin_txid:
  210. if not tx.comment:
  211. lines.append('-') # keep old tx files backwards compatible
  212. lines.append(tx.coin_txid)
  213. if tx.sent_timestamp:
  214. lines.append(f'Sent {tx.sent_timestamp}')
  215. return '\n'.join([make_chksum_6(' '.join(lines))] + lines) + '\n'
  216. def format_data_json():
  217. data = json_dumps({
  218. 'coin_id': coin_id
  219. } | {
  220. k: getattr(tx, k) for k in self.attrs
  221. } | {
  222. 'inputs': [e._asdict() for e in tx.inputs],
  223. 'outputs': [e._asdict() for e in tx.outputs]
  224. } | {
  225. k: getattr(tx, k) for k in self.extra_attrs if getattr(tx, k)
  226. })
  227. return '{{"{}":{},"chksum":"{}"}}'.format(self.data_label, data, make_chksum_6(data))
  228. fmt_data = {'json': format_data_json, 'legacy': format_data_legacy}[tx.file_format]()
  229. if len(fmt_data) > tx.cfg.max_tx_file_size:
  230. die( 'MaxFileSizeExceeded', f'Transaction file size exceeds limit ({tx.cfg.max_tx_file_size} bytes)' )
  231. return fmt_data
  232. def write(self,
  233. add_desc = '',
  234. outdir = None,
  235. ask_write = True,
  236. ask_write_default_yes = False,
  237. ask_tty = True,
  238. ask_overwrite = True ):
  239. if ask_write is False:
  240. ask_write_default_yes = True
  241. if not self.filename:
  242. self.filename = self.make_filename()
  243. if not self.fmt_data:
  244. self.fmt_data = self.format()
  245. from ..fileutil import write_data_to_file
  246. write_data_to_file(
  247. cfg = self.tx.cfg,
  248. outfile = os.path.join((outdir or ''), self.filename),
  249. data = self.fmt_data,
  250. desc = self.tx.desc + add_desc,
  251. ask_overwrite = ask_overwrite,
  252. ask_write = ask_write,
  253. ask_tty = ask_tty,
  254. ask_write_default_yes = ask_write_default_yes,
  255. ignore_opt_outdir = outdir)
  256. @classmethod
  257. def get_proto(cls,cfg,filename,quiet_open=False):
  258. from . import BaseTX
  259. tmp_tx = BaseTX(cfg=cfg)
  260. cls(tmp_tx).parse(filename,metadata_only=True,quiet_open=quiet_open)
  261. return tmp_tx.proto