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

@@ -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)

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)})"

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)