Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 1 addition & 8 deletions electrum/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ class GuiImportError(ImportError):
from .plugin import BasePlugin
from .commands import Commands, known_commands
from .logging import get_logger
from . import crandom # this initializes our RNG state and checks os.urandom is not trivially broken


__version__ = ELECTRUM_VERSION
Expand All @@ -45,11 +46,3 @@ class GuiImportError(ImportError):
pass
else:
raise ImportError("Running with asserts disabled. Refusing to continue. Exiting...")


# Check that os.urandom works
import zlib
length = len(zlib.compress(os.urandom(1000)))
if length <= 900:
raise ImportError("Broken PRNG. Refusing to continue. Exiting...")

3 changes: 2 additions & 1 deletion electrum/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@
from . import crypto
from . import constants
from . import descriptor
from . import crandom

if TYPE_CHECKING:
from .network import Network
Expand Down Expand Up @@ -2311,7 +2312,7 @@ async def get_blinded_path_via(self, node_id: str, dummy_hops: int = 0, wallet:
assert peer, 'node_id not a peer'

path = [pubkey, wallet.lnworker.node_keypair.pubkey]
session_key = os.urandom(32)
session_key = crandom.get_rand_bytes(32)
blinded_path = create_blinded_path(session_key, path=path, final_recipient_data={}, dummy_hops=dummy_hops)

with io.BytesIO() as blinded_path_fd:
Expand Down
153 changes: 153 additions & 0 deletions electrum/crandom.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
# Copyright (C) 2026 The Electrum developers
# Distributed under the MIT software license, see the accompanying
# file LICENCE or http://www.opensource.org/licenses/mit-license.php
#
# Cryptographically secure RNG.
#
# This mostly uses os.urandom, and extreme care should be taken here not to make things worse
# compared to just directly using that.
# We check os.urandom is not trivially broken (passes the zlib test), in which case we panic and runtime exit.
# However, os.urandom could still be subtly "broken" (undetected by us) and produce bad quality output.
# That's the motivation of all this code. We expect os.urandom to work well, BUT if it undetectably does not,
# hopefully mixing in other sources of entropy mitigates the situation somewhat.
#
# inspired by https://github.com/bitcoin/bitcoin/blob/67efced1fc83a0b7215cc1513e7c4754fee0f12f/src/random.h#L25
#
# The logic is split across two modules: crandom.py and crandom_env.py.
# - The core sensitive logic (RNG mixing, extracting random bytes) is in this module (crandom.py),
# which is absolutely security critical and is kept concise to ease review.
# - crandom_env.py contains secondary sources of entropy and potentially platform-specific code.
# Even if all the entropy sources listed in crandom_env.py are broken, assuming os.urandom()
# produces high quality random, this module should never produce low-quality random output.

import hashlib
import os
import threading
from typing import Callable
import zlib

from . import crandom_env


# Check that os.urandom works
length = len(zlib.compress(os.urandom(1000)))
if length <= 900:
raise ImportError("Broken PRNG. Refusing to continue. Exiting...")


def sha512(x: bytes) -> bytes:
assert isinstance(x, bytes)
return hashlib.sha512(x).digest()


CRANDOM_FEEDER_API = Callable[[bytes | str | int], None]


class RNGState:

def __init__(self):
self.lock = threading.Lock()
self._state = os.urandom(32) # secret! access needs lock.
# gather ghetto-entropy:
self.rand_add_refresh() # clock
crandom_env.rand_add_static_env(self.feed_entropy)
self.rand_add_refresh() # clock again

def rand_add_refresh(self) -> None:
"""Gather dynamic environment data that changes over time and mix it in.
This includes a high-precision clock.

Never raises.
"""
crandom_env.rand_add_dynamic_env(self.feed_entropy)

def feed_entropy(self, data: bytes | str | int) -> None:
"""Mix in some data into our internal RNG state, in hopes of increasing entropy.

We MUST be robust for given 'data' not to contain any randomness, it could even be static.
Assuming our internal hash function is cryptographically secure, our internal state
MUST not be left with less entropy than before the call.

Never raises (assuming input type-checks).
Matches CRANDOM_FEEDER_API.
"""
if not data:
return
if isinstance(data, int):
data = hex(data)
if isinstance(data, str):
# we must not raise UnicodeError, hence "backslashreplace"
data = data.encode("utf-8", errors='backslashreplace')
with self.lock:
self._state = sha512(data + self._state)[0:32]
assert len(self._state) == 32

def _mix_extract(self) -> bytes:
"""Return 32 bytes of secure randomness.

Mix in some new entropy from os.urandom first, and then extract 32 bytes.
When mixing in new entropy, H = SHA512(new_entropy || old_rng_state) is computed, and
the first 32 bytes of H are produced as output, while the last 32 bytes
become the new RNG state.

Never raises.
"""
with self.lock:
fresh_entropy = os.urandom(32)
h = sha512(fresh_entropy + self._state)
out, self._state = h[0:32], h[32:64]
assert len(out) == 32
assert len(self._state) == 32
return out

def get_rand_bytes(self, nbytes: int) -> bytes:
"""Returns uniformly distributed bytes, of length nbytes.

Never raises.
"""
assert nbytes >= 0, nbytes
out = b""
while len(out) < nbytes:
out += self._mix_extract()
return out[:nbytes]

def _get_rand_bits(self, nbits: int) -> int:
"""Return a uniformly distributed int in the range [0, 2**nbits).

Never raises.
"""
assert nbits >= 0, nbits
nbytes = nbits // 8 + (1 if nbits % 8 else 0)
rb = self.get_rand_bytes(nbytes)
ri = int.from_bytes(rb, byteorder="big", signed=False)
# strip excess bits (we got up to 7 more than we asked for)
extra_bits = 8 * nbytes - nbits
assert 0 <= extra_bits < 8
ri = ri >> extra_bits
return ri

def get_rand_below(self, upper_bound: int) -> int:
"""Return a uniformly distributed int in the range [0, upper_bound).

Never raises.
"""
assert upper_bound > 0, upper_bound
nbits = upper_bound.bit_length()
ri = upper_bound + 1
# Keep generating random ints until we get one inside the requested range.
# On average, we expect around 2 iterations.
while ri >= upper_bound:
ri = self._get_rand_bits(nbits)
return ri


_rng = RNGState()


########################################
# External API (thread-safe):

get_rand_bytes = _rng.get_rand_bytes
get_rand_below = _rng.get_rand_below
feed_entropy = _rng.feed_entropy
rand_add_refresh = _rng.rand_add_refresh
91 changes: 91 additions & 0 deletions electrum/crandom_env.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
# Copyright (C) 2026 The Electrum developers
# Distributed under the MIT software license, see the accompanying
# file LICENCE or http://www.opensource.org/licenses/mit-license.php
#
# This module is a companion to crandom.py and is only intended to be accessed from there.

import os
import platform
import socket
import ssl
import sys
import threading
import time
from typing import TYPE_CHECKING

from .logging import get_logger

if TYPE_CHECKING:
from .crandom import CRANDOM_FEEDER_API


_logger = get_logger(__name__)


def rand_add_static_env(feed: 'CRANDOM_FEEDER_API') -> None:
"""Gather non-cryptographic environment data that does not change over time
and feed it into feed().
"""
# os
feed(str(os.environ))
feed(getattr(os, "ctermid", lambda: "")())
feed(os.getcwd())
feed(str(os.get_exec_path()))
feed(str(os.getgroups()))
try:
feed(os.getlogin())
except (AttributeError, OSError):
pass
feed(getattr(os, "getpgrp", lambda: "")())
feed(os.getpid())
feed(getattr(os, "getppid", lambda: "")())
feed(str(getattr(os, "getresuid", lambda: "")()))
feed(str(getattr(os, "getresgid", lambda: "")()))
feed(str(getattr(os, "uname", lambda: "")()))
# timezone
feed(time.timezone)
feed(str(time.tzname))
# system locale
import locale
feed(str(locale.getlocale()))
# mac address
from uuid import getnode as get_mac_address
feed(get_mac_address())
# hostname
feed(getattr(socket, "gethostname", lambda: "")())
# threading
feed(getattr(threading, "get_native_id", lambda: "")())
feed(str(threading.enumerate()))
# platform
from .logging import describe_os_version
feed(sys.version)
feed(platform.platform())
feed(describe_os_version())
# version of electrum
from . import ELECTRUM_VERSION
from .logging import get_git_version
feed(ELECTRUM_VERSION)
feed(get_git_version() or "")
# path to this file
feed(__file__)
# memory locations
feed(id(__file__))
feed(id(id))
feed(id(os))
feed(id(feed))
feed(id(ELECTRUM_VERSION))
feed(id(_logger))
feed(id("longish_string_literal"))
feed(id(0))

def rand_add_dynamic_env(feed: 'CRANDOM_FEEDER_API') -> None:
"""Gather non-cryptographic environment data that changes over time and feed it into feed()."""
# time
feed(time.time_ns())
feed(time.process_time_ns())
feed(time.perf_counter_ns())
# openssl
try:
feed(ssl.RAND_bytes(32))
except ssl.SSLError as e:
_logger.info(f"failed to get randomness from ssl: {e!r}")
3 changes: 2 additions & 1 deletion electrum/crypto.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
from .util import assert_bytes, InvalidPassword, to_bytes, to_string, WalletFileException, versiontuple
from .i18n import _
from .logging import get_logger
from . import crandom

_logger = get_logger(__name__)

Expand Down Expand Up @@ -175,7 +176,7 @@ def aes_decrypt_with_iv(key: bytes, iv: bytes, data: bytes) -> bytes:

def EncodeAES_bytes(secret: bytes, msg: bytes) -> bytes:
assert_bytes(msg)
iv = bytes(os.urandom(16))
iv = crandom.get_rand_bytes(16)
ct = aes_encrypt_with_iv(secret, iv, msg)
return iv + ct

Expand Down
17 changes: 16 additions & 1 deletion electrum/gui/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,9 @@

from typing import TYPE_CHECKING, Mapping, Optional

from electrum import crandom
from electrum.crandom import CRANDOM_FEEDER_API

if TYPE_CHECKING:
from . import qt
from electrum.simple_config import SimpleConfig
Expand All @@ -20,7 +23,13 @@ def __init__(self, *, config: 'SimpleConfig', daemon: 'Daemon', plugins: 'Plugin
self.plugins = plugins

def main(self) -> None:
raise NotImplementedError()
"""Main entry point to GUI. Normally this launches a GUI event loop and 'blocks' this thread.
The application will start to gracefully exit after this returns.
"""
# Feed clock into crandom (again). This measures how long it took to create the GUI object.
crandom.rand_add_refresh()
# collect some GUI state as well:
self.rand_add_gui_static_env(crandom.feed_entropy)

def stop(self) -> None:
"""Stops the GUI.
Expand All @@ -31,3 +40,9 @@ def stop(self) -> None:
@classmethod
def version_info(cls) -> Mapping[str, Optional[str]]:
return {}

def rand_add_gui_static_env(self, feed: CRANDOM_FEEDER_API) -> None:
"""Gather non-cryptographic environment data, specific to the GUI, and feed that into crandom.
Never raises.
"""
pass
1 change: 1 addition & 0 deletions electrum/gui/qml/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ def close(self):
self.app.quit()

def main(self):
BaseElectrumGui.main(self)
if not self.app._valid:
return

Expand Down
4 changes: 2 additions & 2 deletions electrum/gui/qml/qebiometrics.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
import os
import secrets
from enum import Enum
from typing import Optional, TYPE_CHECKING

from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot, pyqtProperty

from electrum import crandom
from electrum.i18n import _
from electrum.logging import get_logger
from electrum.base_crash_reporter import send_exception_to_crash_reporter
Expand Down Expand Up @@ -81,7 +81,7 @@ def enable(self, unified_wallet_password: str):
The encryption key for the wrap_key is stored in the AndroidKeyStore.
This way the wallet password doesn't have to leave the process.
"""
wrap_key, iv = secrets.token_bytes(32), secrets.token_bytes(16)
wrap_key, iv = crandom.get_rand_bytes(32), crandom.get_rand_bytes(16)
wrapped_wallet_password = aes_encrypt_with_iv(
key=wrap_key,
iv=iv,
Expand Down
12 changes: 12 additions & 0 deletions electrum/gui/qt/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@
from electrum.keystore import load_keystore
from electrum.bip32 import is_xprv
from electrum import constants
from electrum import crandom

from electrum.gui.common_qt.i18n import ElectrumTranslator
from electrum.gui.messages import TERMS_OF_USE_LATEST_VERSION
Expand Down Expand Up @@ -567,6 +568,7 @@ def init_network(self):
self.daemon.start_network()

def main(self):
BaseElectrumGui.main(self)
# setup Ctrl-C handling and tear-down code first, so that user can easily exit whenever
self.app.setQuitOnLastWindowClosed(False) # so _we_ can decide whether to quit
self.app.lastWindowClosed.connect(self._maybe_quit_if_no_windows_open)
Expand Down Expand Up @@ -618,6 +620,16 @@ def do_copy(self, text: str, *, title: str = None) -> None:
# tooltip cannot be displayed immediately when called from a menu; wait 200ms
QTimer.singleShot(200, lambda: QToolTip.showText(QCursor.pos(), message, None))

def rand_add_gui_static_env(self, feed) -> None:
for screen in self.app.screens():
feed(str(screen.serialNumber()))
feed(str(screen.manufacturer()))
feed(str(screen.model()))
feed(str(screen.name()))
feed(str(screen.size()))
feed(str(screen.availableSize()))
feed(str(screen.refreshRate()))
feed(str(screen.logicalDotsPerInch()))

def standalone_exception_dialog(exception: Union[str, BaseException]) -> None:
app = QApplication.instance()
Expand Down
Loading
Loading