diff --git a/electrum/json_db.py b/electrum/json_db.py index 28d1ec141..ef7742b16 100644 --- a/electrum/json_db.py +++ b/electrum/json_db.py @@ -28,14 +28,30 @@ import json from typing import TYPE_CHECKING, Optional, Sequence, List, Union import jsonpatch +import jsonpointer from . import util -from .util import WalletFileException, profiler +from .util import WalletFileException, profiler, sticky_property from .logging import Logger if TYPE_CHECKING: from .storage import WalletStorage + +# We monkeypatch exceptions in the jsonpatch package to ensure they do not contain secrets from the DB. +# We often log exceptions and offer to send them to the crash reporter, so they must not contain secrets. +jsonpointer.JsonPointerException.__str__ = lambda self: """(JPE) 'redacted'""" +jsonpointer.JsonPointerException.__repr__ = lambda self: """""" +setattr(jsonpointer.JsonPointerException, '__cause__', sticky_property(None)) +setattr(jsonpointer.JsonPointerException, '__context__', sticky_property(None)) +setattr(jsonpointer.JsonPointerException, '__suppress_context__', sticky_property(True)) +jsonpatch.JsonPatchException.__str__ = lambda self: """(JPE) 'redacted'""" +jsonpatch.JsonPatchException.__repr__ = lambda self: """""" +setattr(jsonpatch.JsonPatchException, '__cause__', sticky_property(None)) +setattr(jsonpatch.JsonPatchException, '__context__', sticky_property(None)) +setattr(jsonpatch.JsonPatchException, '__suppress_context__', sticky_property(True)) + + def modifier(func): def wrapper(self, *args, **kwargs): with self.lock: diff --git a/electrum/util.py b/electrum/util.py index 7cb38faf8..d266a6a5c 100644 --- a/electrum/util.py +++ b/electrum/util.py @@ -2199,6 +2199,30 @@ class classproperty(property): return self.fget(owner_cls) +def sticky_property(val): + """Creates a 'property' whose value cannot be changed and that cannot be deleted. + Attempts to change the value are silently ignored. + + >>> class C: pass + ... + >>> setattr(C, 'x', sticky_property(3)) + >>> c = C() + >>> c.x + 3 + >>> c.x = 2 + >>> c.x + 3 + >>> del c.x + >>> c.x + 3 + """ + return property( + fget=lambda self: val, + fset=lambda *args, **kwargs: None, + fdel=lambda *args, **kwargs: None, + ) + + def get_running_loop() -> Optional[asyncio.AbstractEventLoop]: """Returns the asyncio event loop that is *running in this thread*, if any.""" try: diff --git a/tests/test_jsondb.py b/tests/test_jsondb.py new file mode 100644 index 000000000..16129e114 --- /dev/null +++ b/tests/test_jsondb.py @@ -0,0 +1,86 @@ +import contextlib +import copy +import traceback + +import jsonpatch +from jsonpatch import JsonPatchException +from jsonpointer import JsonPointerException + +from . import ElectrumTestCase + + +class TestJsonpatch(ElectrumTestCase): + + async def test_op_replace(self): + data1 = {'foo': 'bar', 'numbers': [1, 3, 4, 8], 'dictlevelA1': {'secret1': 2, 'secret2': 4, 'secret3': 6}} + patches = [{"op": "replace", "path": "/dictlevelA1/secret2", "value": 2222}] + jpatch = jsonpatch.JsonPatch(patches) + data2 = jpatch.apply(data1) + self.assertEqual( + {'foo': 'bar', 'numbers': [1, 3, 4, 8], 'dictlevelA1': {'secret1': 2, 'secret2': 2222, 'secret3': 6}}, + data2 + ) + + @contextlib.contextmanager + def _customAssertRaises(self, *args, **kwargs): + with self.assertRaises(*args, **kwargs) as ctx: + try: + yield ctx + except Exception as e: + # save original traceback now, as assertRaises will destroy most of it imminently: + ctx._customctx_original_tb = "".join(traceback.format_exception(e)) + raise + + async def test_patch_does_not_leak_privatekeys(self): + data1 = { + 'dictlevelB1': 'secret77', + 'dictlevelC1': [1, "secret99", 4, 8], + 'dictlevelA1': {"dictlevelA2_aa": "secret11", "dictlevelA2_bb": "secret12", "dictlevelA2_cc": "secret13"}} + def fail_if_leaking_secret(ctx) -> None: + self.assertNotIn("secret", str(ctx.exception)) + self.assertNotIn("secret", repr(ctx.exception)) + self.assertNotIn("secret", ctx._customctx_original_tb) + self.assertNotIn("dictlevel", str(ctx.exception)) + self.assertNotIn("dictlevel", repr(ctx.exception)) + self.assertNotIn("dictlevel", ctx._customctx_original_tb) + self.assertIn("redacted", str(ctx.exception)) # injected by our monkeypatching + self.assertIn("redacted", repr(ctx.exception)) # injected by our monkeypatching + # op "replace" + with self.subTest(msg="replace_dict_inner_key_missing"): + patches = [{"op": "replace", "path": "/dictlevelA1/dictlevelX2", "value": "nakamoto_secret"}] + jpatch = jsonpatch.JsonPatch(patches) + with self._customAssertRaises(JsonPatchException) as ctx: + data2 = jpatch.apply(data1) + fail_if_leaking_secret(ctx) + with self.subTest(msg="replace_dict_outer_key_missing"): + patches = [{"op": "replace", "path": "/dictlevelX1/dictlevelX2", "value": "nakamoto_secret"}] + jpatch = jsonpatch.JsonPatch(patches) + with self._customAssertRaises(JsonPointerException) as ctx: + data2 = jpatch.apply(data1) + fail_if_leaking_secret(ctx) + # op "remove" + with self.subTest(msg="remove_dict_inner_key_missing"): + patches = [{"op": "remove", "path": "/dictlevelA1/dictlevelX2"}] + jpatch = jsonpatch.JsonPatch(patches) + with self._customAssertRaises(JsonPatchException) as ctx: + data2 = jpatch.apply(data1) + fail_if_leaking_secret(ctx) + with self.subTest(msg="remove_dict_outer_key_missing"): + patches = [{"op": "remove", "path": "/dictlevelX1/dictlevelX2"}] + jpatch = jsonpatch.JsonPatch(patches) + with self._customAssertRaises(JsonPointerException) as ctx: + data2 = jpatch.apply(data1) + fail_if_leaking_secret(ctx) + # op "add" + with self.subTest(msg="add_dict_inner_key_missing"): + patches = [{"op": "add", "path": "/dictlevelA1/dictlevelX2/dictlevelX3/dictlevelX4", "value": "nakamoto_secret"}] + jpatch = jsonpatch.JsonPatch(patches) + with self._customAssertRaises(JsonPointerException) as ctx: + data2 = jpatch.apply(data1) + fail_if_leaking_secret(ctx) + with self.subTest(msg="add_dict_outer_key_missing"): + patches = [{"op": "add", "path": "/dictlevelX1/dictlevelX2/dictlevelX3/dictlevelX4", "value": "nakamoto_secret"}] + jpatch = jsonpatch.JsonPatch(patches) + with self._customAssertRaises(JsonPointerException) as ctx: + data2 = jpatch.apply(data1) + fail_if_leaking_secret(ctx)