From 36efae38757ff37a3ac29c55e9a7633c77424840 Mon Sep 17 00:00:00 2001 From: Sander van Grieken Date: Mon, 10 Feb 2025 14:22:50 +0100 Subject: [PATCH] imports, whitespace --- electrum/bitcoin.py | 41 +++++++++++++++++++++++++++++------------ electrum/blockchain.py | 5 +++++ electrum/channel_db.py | 11 ++++++----- electrum/coinchooser.py | 6 ++++-- 4 files changed, 44 insertions(+), 19 deletions(-) diff --git a/electrum/bitcoin.py b/electrum/bitcoin.py index 9994bd665..88c29aebe 100644 --- a/electrum/bitcoin.py +++ b/electrum/bitcoin.py @@ -23,18 +23,16 @@ # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. -import hashlib -from typing import List, Tuple, TYPE_CHECKING, Optional, Union, Sequence, Any +from typing import Tuple, TYPE_CHECKING, Optional, Union, Sequence import enum from enum import IntEnum, Enum import electrum_ecc as ecc from .util import bfh, BitcoinException, assert_bytes, to_bytes, inv_dict, is_hex_str, classproperty -from . import version from . import segwit_addr from . import constants -from .crypto import sha256d, sha256, hash_160, hmac_oneshot +from .crypto import sha256d, sha256, hash_160 if TYPE_CHECKING: from .network import Network @@ -375,29 +373,36 @@ def hash160_to_p2pkh(h160: bytes, *, net=None) -> str: if net is None: net = constants.net return hash160_to_b58_address(h160, net.ADDRTYPE_P2PKH) + def hash160_to_p2sh(h160: bytes, *, net=None) -> str: if net is None: net = constants.net return hash160_to_b58_address(h160, net.ADDRTYPE_P2SH) + def public_key_to_p2pkh(public_key: bytes, *, net=None) -> str: return hash160_to_p2pkh(hash_160(public_key), net=net) + def hash_to_segwit_addr(h: bytes, witver: int, *, net=None) -> str: if net is None: net = constants.net addr = segwit_addr.encode_segwit_address(net.SEGWIT_HRP, witver, h) assert addr is not None return addr + def public_key_to_p2wpkh(public_key: bytes, *, net=None) -> str: return hash_to_segwit_addr(hash_160(public_key), witver=0, net=net) + def script_to_p2wsh(script: bytes, *, net=None) -> str: return hash_to_segwit_addr(sha256(script), witver=0, net=net) + def p2wsh_nested_script(witness_script: bytes) -> bytes: wsh = sha256(witness_script) return construct_script([0, wsh]) + def pubkey_to_address(txin_type: str, pubkey: str, *, net=None) -> str: from . import descriptor desc = descriptor.get_singlesig_descriptor_from_legacy_leaf(pubkey=pubkey, script_type=txin_type) @@ -593,12 +598,12 @@ def DecodeBase58Check(psz: Union[bytes, str]) -> bytes: # extended WIF for segwit (used in 3.0.x; but still used internally) # the keys in this dict should be a superset of what Imported Wallets can import WIF_SCRIPT_TYPES = { - 'p2pkh':0, - 'p2wpkh':1, - 'p2wpkh-p2sh':2, - 'p2sh':5, - 'p2wsh':6, - 'p2wsh-p2sh':7 + 'p2pkh': 0, + 'p2wpkh': 1, + 'p2wpkh-p2sh': 2, + 'p2sh': 5, + 'p2wsh': 6, + 'p2wsh-p2sh': 7 } WIF_SCRIPT_TYPES_INV = inv_dict(WIF_SCRIPT_TYPES) @@ -679,6 +684,7 @@ def address_from_private_key(sec: str) -> str: public_key = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed) return pubkey_to_address(txin_type, public_key) + def is_segwit_address(addr: str, *, net=None) -> bool: if net is None: net = constants.net try: @@ -687,6 +693,7 @@ def is_segwit_address(addr: str, *, net=None) -> bool: return False return witprog is not None + def is_taproot_address(addr: str, *, net=None) -> bool: if net is None: net = constants.net try: @@ -695,6 +702,7 @@ def is_taproot_address(addr: str, *, net=None) -> bool: return False return witver == 1 + def is_b58_address(addr: str, *, net=None) -> bool: if net is None: net = constants.net try: @@ -706,6 +714,7 @@ def is_b58_address(addr: str, *, net=None) -> bool: return False return True + def is_address(addr: str, *, net=None) -> bool: return is_segwit_address(addr, net=net) \ or is_b58_address(addr, net=net) @@ -733,6 +742,7 @@ def is_minikey(text: str) -> bool: and all(ord(c) in __b58chars for c in text) and sha256(text + '?')[0] == 0x00) + def minikey_to_private_key(text: str) -> bytes: return sha256(text) @@ -740,7 +750,10 @@ def minikey_to_private_key(text: str) -> bytes: def _get_dummy_address(purpose: str) -> str: return redeem_script_to_address('p2wsh', sha256(bytes(purpose, "utf8"))) + _dummy_addr_funcs = set() + + class DummyAddress: """dummy address for fee estimation of funding tx Use e.g. as: DummyAddress.CHANNEL @@ -799,22 +812,24 @@ def taproot_tweak_seckey(seckey0: bytes, h: bytes) -> bytes: TapTreeLeaf = Tuple[int, bytes] TapTree = Union[TapTreeLeaf, Sequence['TapTree']] + # FIXME just use electrum_ecc.util.bip340_tagged_hash instead def bip340_tagged_hash(tag: bytes, msg: bytes) -> bytes: # note: _libsecp256k1.secp256k1_tagged_sha256 benchmarks about 70% slower than this (on my machine) return sha256(sha256(tag) + sha256(tag) + msg) + def taproot_tree_helper(script_tree: TapTree): if isinstance(script_tree, tuple): leaf_version, script = script_tree h = bip340_tagged_hash(b"TapLeaf", bytes([leaf_version]) + witness_push(script)) - return ([((leaf_version, script), bytes())], h) + return [((leaf_version, script), bytes())], h left, left_h = taproot_tree_helper(script_tree[0]) right, right_h = taproot_tree_helper(script_tree[1]) ret = [(l, c + right_h) for l, c in left] + [(l, c + left_h) for l, c in right] if right_h < left_h: left_h, right_h = right_h, left_h - return (ret, bip340_tagged_hash(b"TapBranch", left_h + right_h)) + return ret, bip340_tagged_hash(b"TapBranch", left_h + right_h) def taproot_output_script(internal_pubkey: bytes, *, script_tree: Optional[TapTree]) -> bytes: @@ -850,11 +865,13 @@ def usermessage_magic(message: bytes) -> bytes: length = var_int(len(message)) return b"\x18Bitcoin Signed Message:\n" + length + message + def ecdsa_sign_usermessage(ec_privkey, message: Union[bytes, str], *, is_compressed: bool) -> bytes: message = to_bytes(message, 'utf8') msg32 = sha256d(usermessage_magic(message)) return ec_privkey.ecdsa_sign_recoverable(msg32, is_compressed=is_compressed) + def verify_usermessage_with_address(address: str, sig65: bytes, message: bytes, *, net=None) -> bool: from electrum_ecc import ECPubkey assert_bytes(sig65, message) diff --git a/electrum/blockchain.py b/electrum/blockchain.py index 65b83f554..6264fa4dc 100644 --- a/electrum/blockchain.py +++ b/electrum/blockchain.py @@ -46,9 +46,11 @@ MAX_TARGET = 0x00000000ffffffffffffffffffffffffffffffffffffffffffffffffffffffff class MissingHeader(Exception): pass + class InvalidHeader(Exception): pass + def serialize_header(header_dict: dict) -> bytes: s = ( int.to_bytes(header_dict['version'], length=4, byteorder="little", signed=False) @@ -59,6 +61,7 @@ def serialize_header(header_dict: dict) -> bytes: + int.to_bytes(int(header_dict['nonce']), length=4, byteorder="little", signed=False)) return s + def deserialize_header(s: bytes, height: int) -> dict: if not s: raise InvalidHeader('Invalid header: {}'.format(s)) @@ -74,6 +77,7 @@ def deserialize_header(s: bytes, height: int) -> dict: h['block_height'] = height return h + def hash_header(header: dict) -> str: if header is None: return '0' * 64 @@ -161,6 +165,7 @@ def read_blockchains(config: 'SimpleConfig'): def get_best_chain() -> 'Blockchain': return blockchains[constants.net.GENESIS] + # block hash -> chain work; up to and including that block _CHAINWORK_CACHE = { "0000000000000000000000000000000000000000000000000000000000000000": 0, # virtual block at height -1 diff --git a/electrum/channel_db.py b/electrum/channel_db.py index c589e1323..cc81be62c 100644 --- a/electrum/channel_db.py +++ b/electrum/channel_db.py @@ -37,16 +37,13 @@ from enum import IntEnum import functools from aiorpcx import NetAddress -import electrum_ecc as ecc from electrum_ecc import ECPubkey from .sql_db import SqlDB, sql from . import constants, util from .util import profiler, get_headers_dir, is_ip_address, json_normalize, UserFacingException, is_private_netaddress -from .logging import Logger from .lntransport import LNPeerAddr -from .lnutil import (format_short_channel_id, ShortChannelID, - validate_features, IncompatibleOrInsaneFeatures, InvalidGossipMsg) +from .lnutil import ShortChannelID, validate_features, IncompatibleOrInsaneFeatures, InvalidGossipMsg from .lnverifier import LNChannelVerifier, verify_sig_for_channel_update from .lnmsg import decode_msg from .crypto import sha256d @@ -238,6 +235,7 @@ class NodeInfo(NamedTuple): @staticmethod def parse_addresses_field(addresses_field): buf = addresses_field + def read(n): nonlocal buf data, buf = buf[0:n], buf[n:] @@ -286,6 +284,7 @@ class UpdateStatus(IntEnum): UNCHANGED = 3 GOOD = 4 + class CategorizedChannelUpdates(NamedTuple): orphaned: List # no channel announcement for channel update expired: List # update older than two weeks @@ -303,6 +302,7 @@ def get_mychannel_info(short_channel_id: ShortChannelID, ci = ChannelInfo.from_raw_msg(raw_msg) return ci._replace(capacity_sat=chan.constraints.capacity) + def get_mychannel_policy(short_channel_id: bytes, node_id: bytes, my_channels: Dict[ShortChannelID, 'Channel']) -> Optional[Policy]: chan = my_channels.get(short_channel_id) # type: Optional[Channel] @@ -583,7 +583,6 @@ class ChannelDB(SqlDB): unchanged=unchanged, good=good) - def create_database(self): c = self.conn.cursor() c.execute(create_node_info) @@ -802,6 +801,7 @@ class ChannelDB(SqlDB): def load_data(self): if self.data_loaded.is_set(): return + # Note: this method takes several seconds... mostly due to lnmsg.decode_msg being slow. def maybe_abort(): if self.stopping: @@ -817,6 +817,7 @@ class ChannelDB(SqlDB): except Exception: continue self._addresses[node_id][net_addr] = int(timestamp or 0) + def newest_ts_for_node_id(node_id): newest_ts = 0 for addr, ts in self._addresses[node_id].items(): diff --git a/electrum/coinchooser.py b/electrum/coinchooser.py index e50306445..f75413d37 100644 --- a/electrum/coinchooser.py +++ b/electrum/coinchooser.py @@ -24,11 +24,11 @@ # SOFTWARE. from collections import defaultdict from math import floor, log10 -from typing import NamedTuple, List, Callable, Sequence, Union, Dict, Tuple, Mapping, Type, TYPE_CHECKING +from typing import NamedTuple, List, Callable, Sequence, Dict, Tuple, Mapping, Type, TYPE_CHECKING from decimal import Decimal from .bitcoin import sha256, COIN, is_address -from .transaction import Transaction, TxOutput, PartialTransaction, PartialTxInput, PartialTxOutput +from .transaction import Transaction, PartialTransaction, PartialTxInput, PartialTxOutput from .util import NotEnoughFunds from .logging import Logger @@ -503,12 +503,14 @@ COIN_CHOOSERS = { 'Privacy': CoinChooserPrivacy, } # type: Mapping[str, Type[CoinChooserBase]] + def get_name(config: 'SimpleConfig') -> str: kind = config.WALLET_COIN_CHOOSER_POLICY if kind not in COIN_CHOOSERS: kind = config.cv.WALLET_COIN_CHOOSER_POLICY.get_default_value() return kind + def get_coin_chooser(config: 'SimpleConfig') -> CoinChooserBase: klass = COIN_CHOOSERS[get_name(config)] # note: we enable enable_output_value_rounding by default as