mmgen/mmgen-wallet
1
0
Fork 0
Code Issues Pull requests Projects Releases 28 Packages Wiki Activity

eth: sign transactions with libsecp256k1

Browse source
This commit is contained in:
The MMGen Project 2025-06-29 14:04:47 +00:00
commit 60ca7a2918
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
18 changed files with 48 additions and 436 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/ruff.yaml vendored
View file

@ -41,7 +41,7 @@ jobs:
python3 -m pip install pip setuptools build wheel
python3 -m pip install gmpy2 cryptography pynacl ecdsa aiohttp requests pexpect scrypt semantic-version
python3 -m pip install pycryptodomex pysocks pycoin ipaddress varint ruff
python3 -m pip install py_ecc mypy_extensions monero eth-keys
python3 -m pip install lxml py-ecc monero eth-keys
python3 setup.py build_ext --inplace
- name: Check the code with Ruff static code analyzer

1
MANIFEST.in
View file

@ -4,6 +4,7 @@ include doc/release-notes/*
include doc/wiki/*/*
include examples/*
include extmod/*
include mmgen/proto/eth/*/LICENSE
include mmgen/data/*

4
eth-requirements.txt
View file

@ -1,4 +0,0 @@
# Install with --no-deps. Otherwise, many unneeded dependencies will be
# installed.
py_ecc

2
mmgen/data/version
View file

@ -1 +1 @@
15.1.dev49
15.1.dev50

2
mmgen/proto/eth/contract.py
View file

@ -102,7 +102,7 @@ class Contract:
async def txsign(self, tx_in, key, from_addr, *, chain_id=None):
from .pyethereum.transactions import Transaction
from .tx.transaction import Transaction
if chain_id is None:
res = await self.rpc.call('eth_chainId')

216
mmgen/proto/eth/pyethereum/transactions.py
View file

@ -1,216 +0,0 @@
#
# Adapted from: # https://github.com/ethereum/pyethereum/blob/master/ethereum/transactions.py
#
from .. import rlp
from ..rlp.sedes import big_endian_int,binary
from .utils import (
str_to_bytes,encode_hex,ascii_chr,big_endian_to_int,
TT256,mk_contract_address,
ecsign,ecrecover_to_pub,normalize_key )
from . import utils
class InvalidTransaction(Exception): pass
class opcodes(object):
GTXCOST = 21000 # TX BASE GAS COST
GTXDATAZERO = 4 # TX DATA ZERO BYTE GAS COST
GTXDATANONZERO = 68 # TX DATA NON ZERO BYTE GAS COST
# in the yellow paper it is specified that s should be smaller than
# secpk1n (eq.205)
secpk1n = 115792089237316195423570985008687907852837564279074904382605163141518161494337
null_address = b'\xff' * 20
class Transaction(rlp.Serializable):
"""
A transaction is stored as:
[nonce, gasprice, startgas, to, value, data, v, r, s]
nonce is the number of transactions already sent by that account, encoded
in binary form (eg. 0 -> '', 7 -> '\x07', 1000 -> '\x03\xd8').
(v,r,s) is the raw Electrum-style signature of the transaction without the
signature made with the private key corresponding to the sending account,
with 0 <= v <= 3. From an Electrum-style signature (65 bytes) it is
possible to extract the public key, and thereby the address, directly.
A valid transaction is one where:
(i) the signature is well-formed (ie. 0 <= v <= 3, 0 <= r < P, 0 <= s < N,
0 <= r < P - N if v >= 2), and
(ii) the sending account has enough funds to pay the fee and the value.
"""
fields = [
('nonce', big_endian_int),
('gasprice', big_endian_int),
('startgas', big_endian_int),
('to', utils.address),
('value', big_endian_int),
('data', binary),
('v', big_endian_int),
('r', big_endian_int),
('s', big_endian_int),
]
_sender = None
def __init__(self, nonce, gasprice, startgas, to, value, data, v=0, r=0, s=0):
# self.data = None
to = utils.normalize_address(to, allow_blank=True)
super(
Transaction,
self).__init__(
nonce,
gasprice,
startgas,
to,
value,
data,
v,
r,
s)
if self.gasprice >= TT256 or self.startgas >= TT256 or \
self.value >= TT256 or self.nonce >= TT256:
raise InvalidTransaction("Values way too high!")
@property
def sender(self):
if not self._sender:
# Determine sender
if self.r == 0 and self.s == 0:
self._sender = null_address
else:
if self.v in (27, 28):
vee = self.v
sighash = utils.sha3(rlp.encode(unsigned_tx_from_tx(self), UnsignedTransaction))
elif self.v >= 37:
vee = self.v - self.network_id * 2 - 8
assert vee in (27, 28)
rlpdata = rlp.encode(rlp.infer_sedes(self).serialize(self)[:-3] + [self.network_id, '', ''])
sighash = utils.sha3(rlpdata)
else:
raise InvalidTransaction("Invalid V value")
if self.r >= secpk1n or self.s >= secpk1n or self.r == 0 or self.s == 0:
raise InvalidTransaction("Invalid signature values!")
pub = ecrecover_to_pub(sighash, vee, self.r, self.s)
if pub == b'\x00' * 64:
raise InvalidTransaction(
"Invalid signature (zero privkey cannot sign)")
self._sender = utils.sha3(pub)[-20:]
return self._sender
@property
def network_id(self):
if self.r == 0 and self.s == 0:
return self.v
elif self.v in (27, 28):
return None
else:
return ((self.v - 1) // 2) - 17
@sender.setter
def sender(self, value):
self._sender = value
def sign(self, key, network_id=None):
"""Sign this transaction with a private key.
A potentially already existing signature would be overridden.
"""
if network_id is None:
rawhash = utils.sha3(rlp.encode(unsigned_tx_from_tx(self), UnsignedTransaction))
else:
assert 1 <= network_id < 2**63 - 18
rlpdata = rlp.encode(rlp.infer_sedes(self).serialize(self)[:-3] + [network_id, b'', b''])
rawhash = utils.sha3(rlpdata)
key = normalize_key(key)
v, r, s = ecsign(rawhash, key)
if network_id is not None:
v += 8 + network_id * 2
ret = self.copy(
v=v, r=r, s=s
)
ret._sender = utils.privtoaddr(key)
return ret
@property
def hash(self):
return utils.sha3(rlp.encode(self))
def to_dict(self):
d = {}
for name, _ in self.__class__._meta.fields:
d[name] = getattr(self, name)
if name in ('to', 'data'):
d[name] = '0x' + encode_hex(d[name])
d['sender'] = '0x' + encode_hex(self.sender)
d['hash'] = '0x' + encode_hex(self.hash)
return d
@property
def intrinsic_gas_used(self):
num_zero_bytes = str_to_bytes(self.data).count(ascii_chr(0))
num_non_zero_bytes = len(self.data) - num_zero_bytes
return (opcodes.GTXCOST
# + (0 if self.to else opcodes.CREATE[3])
+ opcodes.GTXDATAZERO * num_zero_bytes
+ opcodes.GTXDATANONZERO * num_non_zero_bytes)
@property
def creates(self):
"returns the address of a contract created by this tx"
if self.to in (b'', '\0' * 20):
return mk_contract_address(self.sender, self.nonce)
def __eq__(self, other):
return isinstance(other, self.__class__) and self.hash == other.hash
def __lt__(self, other):
return isinstance(other, self.__class__) and self.hash < other.hash
def __hash__(self):
return utils.big_endian_to_int(self.hash)
def __ne__(self, other):
return not self.__eq__(other)
def __repr__(self):
return '<Transaction(%s)>' % encode_hex(self.hash)[:4]
def __structlog__(self):
return encode_hex(self.hash)
# This method should be called for block numbers >= HOMESTEAD_FORK_BLKNUM only.
# The >= operator is replaced by > because the integer division N/2 always produces the value
# which is by 0.5 less than the real N/2
def check_low_s_metropolis(self):
if self.s > secpk1n // 2:
raise InvalidTransaction("Invalid signature S value!")
def check_low_s_homestead(self):
if self.s > secpk1n // 2 or self.s == 0:
raise InvalidTransaction("Invalid signature S value!")
class UnsignedTransaction(rlp.Serializable):
fields = [
(field, sedes) for field, sedes in Transaction._meta.fields
if field not in "vrs"
]
def unsigned_tx_from_tx(tx):
return UnsignedTransaction(
nonce=tx.nonce,
gasprice=tx.gasprice,
startgas=tx.startgas,
to=tx.to,
value=tx.value,
data=tx.data,
)

206
mmgen/proto/eth/pyethereum/utils.py
View file

@ -1,44 +1,27 @@
#
# Adapted from: https://github.com/ethereum/pyethereum/blob/master/ethereum/utils.py
# only funcs, vars required by vendored rlp module retained
#
from py_ecc.secp256k1 import privtopub,ecdsa_raw_sign,ecdsa_raw_recover
from .. import rlp
from ..rlp.sedes import Binary
import struct, functools
from typing import Any, Callable, TypeVar
from ....util2 import get_keccak
keccak_256 = get_keccak()
ALL_BYTES = tuple(struct.pack('B', i) for i in range(256))
def sha3_256(bstr):
return keccak_256(bstr).digest()
import struct
ALL_BYTES = tuple( struct.pack('B', i) for i in range(256) )
# from eth_utils:
# Type ignored for `codecs.decode()` due to lack of mypy support for 'hex' encoding
# https://github.com/python/typeshed/issues/300
from typing import AnyStr,Any,Callable,TypeVar
import codecs
import functools
T = TypeVar("T")
TVal = TypeVar("TVal")
TKey = TypeVar("TKey")
T = TypeVar('T')
def apply_to_return_value(callback: Callable[..., T]) -> Callable[..., Callable[..., T]]:
def outer(fn):
# We would need to type annotate *args and **kwargs but doing so segfaults
# the PyPy builds. We ignore instead.
@functools.wraps(fn)
def inner(*args, **kwargs) -> T: # type: ignore
return callback(fn(*args, **kwargs))
def outer(fn):
# We would need to type annotate *args and **kwargs but doing so segfaults
# the PyPy builds. We ignore instead.
@functools.wraps(fn)
def inner(*args, **kwargs) -> T: # type: ignore
return callback(fn(*args, **kwargs))
return inner
return inner
return outer
return outer
to_list = apply_to_return_value(list)
to_set = apply_to_return_value(set)
@ -46,168 +29,11 @@ to_dict = apply_to_return_value(dict)
to_tuple = apply_to_return_value(tuple)
to_list = apply_to_return_value(list)
def encode_hex_0x(value: AnyStr) -> str:
if not is_string(value):
raise TypeError("Value must be an instance of str or unicode")
binary_hex = codecs.encode(value, "hex") # type: ignore
return '0x' + binary_hex.decode("ascii")
def decode_hex(value: str) -> bytes:
if not isinstance(value,str):
raise TypeError("Value must be an instance of str")
return codecs.decode(remove_0x_prefix(value), "hex") # type: ignore
def is_bytes(value: Any) -> bool:
return isinstance(value, (bytes,bytearray))
return isinstance(value, (bytes,bytearray))
def int_to_big_endian(value: int) -> bytes:
return value.to_bytes((value.bit_length() + 7) // 8 or 1, "big")
return value.to_bytes((value.bit_length() + 7) // 8 or 1, 'big')
def big_endian_to_int(value: bytes) -> int:
return int.from_bytes(value, "big")
# end from eth_utils
class Memoize:
def __init__(self, fn):
self.fn = fn
self.memo = {}
def __call__(self, *args):
if args not in self.memo:
self.memo[args] = self.fn(*args)
return self.memo[args]
TT256 = 2 ** 256
def is_numeric(x): return isinstance(x, int)
def is_string(x): return isinstance(x, bytes)
def to_string(value):
if isinstance(value, bytes):
return value
if isinstance(value, str):
return bytes(value, 'utf-8')
if isinstance(value, int):
return bytes(str(value), 'utf-8')
unicode = str
def encode_int32(v):
return v.to_bytes(32, byteorder='big')
def str_to_bytes(value):
if isinstance(value, bytearray):
value = bytes(value)
if isinstance(value, bytes):
return value
return bytes(value, 'utf-8')
def ascii_chr(n):
return ALL_BYTES[n]
def encode_hex(n):
if isinstance(n, str):
return encode_hex(n.encode('ascii'))
return encode_hex_0x(n)[2:]
def ecrecover_to_pub(rawhash, v, r, s):
result = ecdsa_raw_recover(rawhash, (v, r, s))
if result:
x, y = result
pub = encode_int32(x) + encode_int32(y)
else:
raise ValueError('Invalid VRS')
assert len(pub) == 64
return pub
def ecsign(rawhash, key):
return ecdsa_raw_sign(rawhash, key)
def mk_contract_address(sender, nonce):
return sha3(rlp.encode([normalize_address(sender), nonce]))[12:]
def mk_metropolis_contract_address(sender, initcode):
return sha3(normalize_address(sender) + initcode)[12:]
def sha3(seed):
return sha3_256(to_string(seed))
assert encode_hex(
sha3(b'')) == 'c5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470'
@Memoize
def privtoaddr(k):
k = normalize_key(k)
x, y = privtopub(k)
return sha3(encode_int32(x) + encode_int32(y))[12:]
def normalize_address(x, allow_blank=False):
if is_numeric(x):
return int_to_addr(x)
if allow_blank and x in {'', b''}:
return b''
if len(x) in (42, 50) and x[:2] in {'0x', b'0x'}:
x = x[2:]
if len(x) in (40, 48):
x = decode_hex(x)
if len(x) == 24:
assert len(x) == 24 and sha3(x[:20])[:4] == x[-4:]
x = x[:20]
if len(x) != 20:
raise Exception("Invalid address format: %r" % x)
return x
def normalize_key(key):
if is_numeric(key):
o = encode_int32(key)
elif len(key) == 32:
o = key
elif len(key) == 64:
o = decode_hex(key)
elif len(key) == 66 and key[:2] == '0x':
o = decode_hex(key[2:])
else:
raise Exception("Invalid key format: %r" % key)
if o == b'\x00' * 32:
raise Exception("Zero privkey invalid")
return o
def int_to_addr(x):
o = [b''] * 20
for i in range(20):
o[19 - i] = ascii_chr(x & 0xff)
x >>= 8
return b''.join(o)
def remove_0x_prefix(s):
return s[2:] if s[:2] in (b'0x', '0x') else s
class Denoms():
def __init__(self):
self.wei = 1
self.babbage = 10 ** 3
self.ada = 10 ** 3
self.kwei = 10 ** 6
self.lovelace = 10 ** 6
self.mwei = 10 ** 6
self.shannon = 10 ** 9
self.gwei = 10 ** 9
self.szabo = 10 ** 12
self.finney = 10 ** 15
self.mether = 10 ** 15
self.ether = 10 ** 18
self.turing = 2 ** 256 - 1
address = Binary.fixed_length(20, allow_empty=True)
return int.from_bytes(value, 'big')

2
mmgen/proto/eth/tx/online.py
View file

@ -80,7 +80,7 @@ class TokenOnlineSigned(TokenSigned, OnlineSigned):
o['amt'] = t.transferdata2amt(o['data'])
o['token_to'] = t.transferdata2sendaddr(o['data'])
if self.is_swap:
from ..pyethereum.transactions import Transaction
from .transaction import Transaction
from .. import rlp
etx = rlp.decode(bytes.fromhex(self.serialized2), Transaction)
d = etx.to_dict()

2
mmgen/proto/eth/tx/signed.py
View file

@ -22,7 +22,7 @@ class Signed(Completed, TxBase.Signed):
desc = 'signed transaction'
def parse_txfile_serialized_data(self):
from ..pyethereum.transactions import Transaction
from .transaction import Transaction
from .. import rlp
etx = rlp.decode(bytes.fromhex(self.serialized), Transaction)
d = etx.to_dict() # ==> hex values have '0x' prefix, 0 is '0x'

2
mmgen/proto/eth/tx/unsigned.py
View file

@ -52,7 +52,7 @@ class Unsigned(VmUnsigned, Completed, TxBase.Unsigned):
'nonce': o['nonce'],
'data': self.swap_memo.encode() if self.is_swap else bytes.fromhex(o['data'])}
from ..pyethereum.transactions import Transaction
from .transaction import Transaction
etx = Transaction(**o_conv).sign(wif, o['chainId'])
assert etx.sender.hex() == o['from'], (
'Sender address recovered from signature does not match true sender')

21
mmgen/proto/eth/util.py
View file

@ -14,6 +14,8 @@ proto.eth.util: various utilities for Ethereum base protocol
from ...util2 import get_keccak
v_base = 27
def decrypt_geth_keystore(cfg, wallet_fn, passwd, *, check_addr=True):
"""
Decrypt the encrypted private key in a Geth keystore wallet, returning the decrypted key
@ -54,9 +56,9 @@ def ec_sign_message_with_privkey(cfg, message, key, msghash_type):
Conforms to the standard defined by the Geth `eth_sign` JSON-RPC call
"""
from py_ecc.secp256k1 import ecdsa_raw_sign
v, r, s = ecdsa_raw_sign(hash_message(cfg, message, msghash_type), key)
return '{:064x}{:064x}{:02x}'.format(r, s, v)
from ..secp256k1.secp256k1 import sign_msghash
sig, recid = sign_msghash(hash_message(cfg, message, msghash_type), key)
return sig.hex() + '{:02x}'.format(v_base + recid)
def ec_recover_pubkey(cfg, message, sig, msghash_type):
"""
@ -65,12 +67,13 @@ def ec_recover_pubkey(cfg, message, sig, msghash_type):
Conforms to the standard defined by the Geth `eth_sign` JSON-RPC call
"""
from py_ecc.secp256k1 import ecdsa_raw_recover
r, s, v = (sig[:64], sig[64:128], sig[128:])
return '{:064x}{:064x}'.format(
*ecdsa_raw_recover(
hash_message(cfg, message, msghash_type), tuple(int(hexstr, 16) for hexstr in (v, r, s)))
)
from ..secp256k1.secp256k1 import pubkey_recover
sig_bytes = bytes.fromhex(sig)
return pubkey_recover(
hash_message(cfg, message, msghash_type),
sig_bytes[:64],
sig_bytes[64] - v_base,
False).hex()
def compute_contract_addr(cfg, deployer_addr, nonce):
from . import rlp

2
mmgen/util2.py
View file

@ -55,7 +55,7 @@ def load_cryptodome(called=[]):
sys.modules['Cryptodome'] = Crypto # Cryptodome == pycryptodomex
called.append(True)
# called with no arguments by pyethereum.utils:
# called with no arguments by proto.eth.tx.transaction:
def get_keccak(cfg=None, cached_ret=[]):
if not cached_ret:

2
nix/user-packages.nix
View file

@ -40,7 +40,7 @@ rec {
python-packages = with python.pkgs; {
# pycryptodome = pycryptodome; # altcoins
# py-ecc = py-ecc; # ETH, ETC
# py-ecc = py-ecc; # test suite
# pysocks = pysocks; # XMR
# monero = monero; # XMR (test suite)
# eth-keys = eth-keys; # ETH, ETC (test suite)

1
test-requirements.txt
View file

@ -1,3 +1,4 @@
pycoin
monero
eth_keys
py_ecc

5
test/cmdtest_d/ethdev.py
View file

@ -1183,8 +1183,9 @@ class CmdTestEthdev(CmdTestEthdevMethods, CmdTestBase, CmdTestShared):
# Compare signatures
imsg(f'Message: {self.message}')
imsg(f'Signature: {sig}')
cmp_or_die(sig, sig_chk, 'message signatures')
imsg('Geth and MMGen signatures match')
if sig != sig_chk:
msg(yellow('Warning: Geth and MMGen signatures don’t match!'))
time.sleep(2)
return 'ok'

7
test/modtest_d/dep.py
View file

@ -18,7 +18,7 @@ from ..include.common import cfg, vmsg, check_solc_ver
class unit_tests:
altcoin_deps = ('py_ecc', 'solc', 'keccak', 'pysocks', 'semantic_version')
altcoin_deps = ('solc', 'keccak', 'pysocks', 'semantic_version')
win_skip = ('led', 'semantic_version')
def secp256k1(self, name, ut):
@ -55,11 +55,6 @@ class unit_tests:
else:
return True
def py_ecc(self, name, ut): # ETH
from py_ecc.secp256k1 import privtopub
privtopub(b'f' * 32)
return True
def pysocks(self, name, ut):
import requests, urllib3
urllib3.disable_warnings()

5
test/modtest_d/testdep.py
View file

@ -67,3 +67,8 @@ class unit_tests:
def ssh_socks_proxy(self, name, ut):
from test.cmdtest_d.include.proxy import TestProxy
return TestProxy(None, cfg)
def py_ecc(self, name, ut):
from py_ecc.secp256k1 import privtopub
privtopub(b'f' * 32)
return True

2
test/test-release.d/cfg.sh
View file

@ -309,7 +309,7 @@ init_tests() {
t $tooltest2_py --coin=rune
- $tooltest2_py --fork # run once with --fork so commands are actually executed
"
[ "$SKIP_ALT_DEP" ] && t_tool2_skip='a e t' # skip ETH,ETC: txview requires py_ecc
[ "$SKIP_ALT_DEP" ] && t_tool2_skip='a e t'
d_tool="'mmgen-tool' utility (all supported coins)"
t_tool="