From bec78b4210f5a5e76e1eca3a44443c2e6c50508a Mon Sep 17 00:00:00 2001 From: SomberNight Date: Mon, 17 Jun 2024 16:24:27 +0000 Subject: [PATCH] util.error_text_str_to_safe_str: truncate long errors The messages are sometimes logged and sometimes shown to the user, - for logging we might not want to truncate or have higher limits, - but when shown to the user, we definitely want to truncate the error text. It is simplest to just do the truncation here, at the lowest level. Note that we usually prepend the error text with a header e.g. "[DO NOT TRUST THIS MESSAGE]" and if the error text is too long, this header at the beginning might get "lost" in some way. Hence we should truncate the error text. --- electrum/lnpeer.py | 4 ++-- electrum/util.py | 19 +++++++++++++++---- tests/test_util.py | 10 +++++++++- 3 files changed, 26 insertions(+), 7 deletions(-) diff --git a/electrum/lnpeer.py b/electrum/lnpeer.py index f2020a2cb..f1639f75b 100644 --- a/electrum/lnpeer.py +++ b/electrum/lnpeer.py @@ -263,7 +263,7 @@ class Peer(Logger): err_bytes = payload['data'] is_known_chan_id = (chan_id in self.channels) or (chan_id in self.temp_id_to_id) self.logger.info(f"remote peer sent warning [DO NOT TRUST THIS MESSAGE]: " - f"{error_text_bytes_to_safe_str(err_bytes)}. chan_id={chan_id.hex()}. " + f"{error_text_bytes_to_safe_str(err_bytes, max_len=None)}. chan_id={chan_id.hex()}. " f"{is_known_chan_id=}") def on_error(self, payload): @@ -271,7 +271,7 @@ class Peer(Logger): err_bytes = payload['data'] is_known_chan_id = (chan_id in self.channels) or (chan_id in self.temp_id_to_id) self.logger.info(f"remote peer sent error [DO NOT TRUST THIS MESSAGE]: " - f"{error_text_bytes_to_safe_str(err_bytes)}. chan_id={chan_id.hex()}. " + f"{error_text_bytes_to_safe_str(err_bytes, max_len=None)}. chan_id={chan_id.hex()}. " f"{is_known_chan_id=}") if chan_id in self.channels: self.schedule_force_closing(chan_id) diff --git a/electrum/util.py b/electrum/util.py index a7da75c75..63369b755 100644 --- a/electrum/util.py +++ b/electrum/util.py @@ -2087,14 +2087,17 @@ def get_running_loop() -> Optional[asyncio.AbstractEventLoop]: return None -def error_text_str_to_safe_str(err: str) -> str: +def error_text_str_to_safe_str(err: str, *, max_len: Optional[int] = 500) -> str: """Converts an untrusted error string to a sane printable ascii str. Never raises. """ - return error_text_bytes_to_safe_str(err.encode("ascii", errors='backslashreplace')) + text = error_text_bytes_to_safe_str( + err.encode("ascii", errors='backslashreplace'), + max_len=None) + return truncate_text(text, max_len=max_len) -def error_text_bytes_to_safe_str(err: bytes) -> str: +def error_text_bytes_to_safe_str(err: bytes, *, max_len: Optional[int] = 500) -> str: """Converts an untrusted error bytes text to a sane printable ascii str. Never raises. @@ -2107,4 +2110,12 @@ def error_text_bytes_to_safe_str(err: bytes) -> str: # convert to ascii, to get rid of unicode stuff ascii_text = err.decode("ascii", errors='backslashreplace') # do repr to handle ascii special chars (especially when printing/logging the str) - return repr(ascii_text) + text = repr(ascii_text) + return truncate_text(text, max_len=max_len) + + +def truncate_text(text: str, *, max_len: Optional[int]) -> str: + if max_len is None or len(text) <= max_len: + return text + else: + return text[:max_len] + f"... (truncated. orig_len={len(text)})" diff --git a/tests/test_util.py b/tests/test_util.py index f64af19d3..59d8d4b99 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -362,7 +362,11 @@ class TestUtil(ElectrumTestCase): util.error_text_bytes_to_safe_str(b'here is some unicode: \xe2\x82\xbf \xf0\x9f\x98\x80 \xf0\x9f\x98\x88')) # not even unicode self.assertEqual("""\'\\x00\\x01\\x02\\x03\\x04\\x05\\x06\\x07\\x08\\t\\n\\x0b\\x0c\\r\\x0e\\x0f\\x10\\x11\\x12\\x13\\x14\\x15\\x16\\x17\\x18\\x19\\x1a\\x1b\\x1c\\x1d\\x1e\\x1f !"#$%&\\\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\\x7f\\\\x80\\\\x81\\\\x82\\\\x83\\\\x84\\\\x85\\\\x86\\\\x87\\\\x88\\\\x89\\\\x8a\\\\x8b\\\\x8c\\\\x8d\\\\x8e\\\\x8f\\\\x90\\\\x91\\\\x92\\\\x93\\\\x94\\\\x95\\\\x96\\\\x97\\\\x98\\\\x99\\\\x9a\\\\x9b\\\\x9c\\\\x9d\\\\x9e\\\\x9f\\\\xa0\\\\xa1\\\\xa2\\\\xa3\\\\xa4\\\\xa5\\\\xa6\\\\xa7\\\\xa8\\\\xa9\\\\xaa\\\\xab\\\\xac\\\\xad\\\\xae\\\\xaf\\\\xb0\\\\xb1\\\\xb2\\\\xb3\\\\xb4\\\\xb5\\\\xb6\\\\xb7\\\\xb8\\\\xb9\\\\xba\\\\xbb\\\\xbc\\\\xbd\\\\xbe\\\\xbf\\\\xc0\\\\xc1\\\\xc2\\\\xc3\\\\xc4\\\\xc5\\\\xc6\\\\xc7\\\\xc8\\\\xc9\\\\xca\\\\xcb\\\\xcc\\\\xcd\\\\xce\\\\xcf\\\\xd0\\\\xd1\\\\xd2\\\\xd3\\\\xd4\\\\xd5\\\\xd6\\\\xd7\\\\xd8\\\\xd9\\\\xda\\\\xdb\\\\xdc\\\\xdd\\\\xde\\\\xdf\\\\xe0\\\\xe1\\\\xe2\\\\xe3\\\\xe4\\\\xe5\\\\xe6\\\\xe7\\\\xe8\\\\xe9\\\\xea\\\\xeb\\\\xec\\\\xed\\\\xee\\\\xef\\\\xf0\\\\xf1\\\\xf2\\\\xf3\\\\xf4\\\\xf5\\\\xf6\\\\xf7\\\\xf8\\\\xf9\\\\xfa\\\\xfb\\\\xfc\\\\xfd\\\\xfe\\\\xff\'""", - util.error_text_bytes_to_safe_str(bytes(range(256)))) + util.error_text_bytes_to_safe_str(bytes(range(256)), max_len=1000)) + # long text + t1 = util.error_text_bytes_to_safe_str(b"test" * 10000) + self.assertTrue(t1.endswith("... (truncated. orig_len=40002)")) + self.assertTrue(len(t1) < 550) def test_error_text_str_to_safe_str(self): # ascii @@ -373,6 +377,10 @@ class TestUtil(ElectrumTestCase): # unicode self.assertEqual("'here is some unicode: \\\\u20bf \\\\U0001f600 \\\\U0001f608'", util.error_text_str_to_safe_str("here is some unicode: ₿ 😀 😈")) + # long text + t1 = util.error_text_str_to_safe_str("test"*10000) + self.assertTrue(t1.endswith("... (truncated. orig_len=40002)")) + self.assertTrue(len(t1) < 550) def test_age(self): now = datetime(2023, 4, 16, 22, 30, 00)