-
Notifications
You must be signed in to change notification settings - Fork 3.5k
add crandom.py, for our RNG, where we mix stuff with os.urandom #10794
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| 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 | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. as far as I understand you pull OS specific values when creating the state |
||
|
|
||
| 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] | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this also applies to other places. I am not sure about taking the first half of SHA512. Thus wouldn't it make more sense to do The - admitedly paranoid - reasoning: The hash is an additional component capable, at least in principle, of damaging a good source if the implementation or construction is defective while xor should always work? |
||
| 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) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. see comment from |
||
| 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 | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. also nbytes needs to by smaller equal than 32 I guess |
||
| 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 | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It absolutely makes sense to expose this API. however I am not sure if users who need this API will understand that they need it and build their application accordingly. See other comment in |
||
| rand_add_refresh = _rng.rand_add_refresh | ||
| 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())) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Seems to be supported only under linux? c.f. https://docs.python.org/3/library/os.html#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}") | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am not sure about this sanity check. I think there is a non zero chance that true random output is compressable. That probability should be small however and I see what you are trying to achieve here but I am not sure if other means are more useful