From 546d570ca27b01963867f37188f9f8c3ceebb05e 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 3f8c35242..b70ff5738 100644 --- a/electrum/lnpeer.py +++ b/electrum/lnpeer.py @@ -259,7 +259,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): @@ -267,7 +267,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 7b17bb68b..b0ab66ac6 100644 --- a/electrum/util.py +++ b/electrum/util.py @@ -2058,14 +2058,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. @@ -2078,4 +2081,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)