From daf950c0a50f852fc720545870764f2e0bed69e2 Mon Sep 17 00:00:00 2001 From: Leon Buchner Date: Fri, 14 Mar 2025 12:26:55 +0100 Subject: [PATCH 1/5] replace kwargs double backslash for multiline messages --- rocketchat_API/APISections/base.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/rocketchat_API/APISections/base.py b/rocketchat_API/APISections/base.py index cb070dd..e39f788 100644 --- a/rocketchat_API/APISections/base.py +++ b/rocketchat_API/APISections/base.py @@ -83,6 +83,12 @@ def call_api_get(self, method, api_path=None, **kwargs): ) def call_api_post(self, method, files=None, use_json=None, **kwargs): + if "text" in kwargs: + kwargs["text"] = kwargs["text"].replace('\\n', '\n') + kwargs["text"] = kwargs["text"].replace('\\r', '\r') + kwargs["text"] = kwargs["text"].replace('\\t', '\t') + kwargs["text"] = kwargs["text"].replace('\\b', '\b') + kwargs["text"] = kwargs["text"].replace('\\f', '\f') reduced_args = self.__reduce_kwargs(kwargs) # Since pass is a reserved word in Python it has to be injected on the request dict # Some methods use pass (users.register) and others password (users.create) From e3e90672645891320921cacac6db47e847b60a13 Mon Sep 17 00:00:00 2001 From: Leon Buchner Date: Thu, 4 Sep 2025 13:07:59 +0200 Subject: [PATCH 2/5] remove text extraction from kwargs --- rocketchat_API/APISections/base.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/rocketchat_API/APISections/base.py b/rocketchat_API/APISections/base.py index 8d7b607..8498fd2 100644 --- a/rocketchat_API/APISections/base.py +++ b/rocketchat_API/APISections/base.py @@ -83,12 +83,6 @@ def call_api_get(self, method, api_path=None, **kwargs): ) def call_api_post(self, method, files=None, use_json=None, **kwargs): - if "text" in kwargs: - kwargs["text"] = kwargs["text"].replace('\\n', '\n') - kwargs["text"] = kwargs["text"].replace('\\r', '\r') - kwargs["text"] = kwargs["text"].replace('\\t', '\t') - kwargs["text"] = kwargs["text"].replace('\\b', '\b') - kwargs["text"] = kwargs["text"].replace('\\f', '\f') reduced_args = self.__reduce_kwargs(kwargs) # Since pass is a reserved word in Python it has to be injected on the request dict # Some methods use pass (users.register) and others password (users.create) From 630d6298abfdb40b7640a2a4465b293caebfd364 Mon Sep 17 00:00:00 2001 From: Leon Buchner Date: Thu, 4 Sep 2025 13:08:21 +0200 Subject: [PATCH 3/5] change chat message --- rocketchat_API/APISections/chat.py | 58 ++++++++++++++++++++++-------- 1 file changed, 44 insertions(+), 14 deletions(-) diff --git a/rocketchat_API/APISections/chat.py b/rocketchat_API/APISections/chat.py index 1df28c0..2d9e9ec 100644 --- a/rocketchat_API/APISections/chat.py +++ b/rocketchat_API/APISections/chat.py @@ -5,21 +5,51 @@ class RocketChatChat(RocketChatBase): def chat_post_message(self, text, room_id=None, channel=None, **kwargs): """Posts a new chat message.""" - if room_id: - if text: - return self.call_api_post( - "chat.postMessage", roomId=room_id, text=text, kwargs=kwargs - ) - return self.call_api_post("chat.postMessage", roomId=room_id, kwargs=kwargs) if channel: - if text: - return self.call_api_post( - "chat.postMessage", channel=channel, text=text, kwargs=kwargs - ) - return self.call_api_post( - "chat.postMessage", channel=channel, kwargs=kwargs - ) - raise RocketMissingParamException("roomId or channel required") + kwargs["channel"] = channel + if room_id: + kwargs["roomId"] = room_id + if not channel or not room_id: + raise RocketMissingParamException("roomId or channel required") + if not text: + raise RocketMissingParamException("text required") + payload = {"text": text, **kwargs} + self.__sanitize_payload_text_fields(payload) + return self.call_api_post("chat.postMessage", channel=channel, kwargs=payload) + + def __sanitize_text(self, value: str) -> str: + """Return value with common double-escaped control sequences normalized.""" + if not isinstance(value, str): + return value + return ( + value.replace("\\n", "\n") + .replace("\\r", "\r") + .replace("\\t", "\t") + .replace("\\b", "\b") + .replace("\\f", "\f") + ) + + def __sanitize_payload_text_fields(self, payload: dict) -> dict: + """ + In-place sanitize of typical text fields in chat payloads: + - payload["text"] + - payload["attachments"][i]["text"] + + Returns the mutated payload for convenience. + """ + if not isinstance(payload, dict): + return payload + + if "text" in payload and isinstance(payload["text"], str): + payload["text"] = self.__sanitize_text(payload["text"]) + + attachments = payload.get("attachments") + if isinstance(attachments, list): + for att in attachments: + if isinstance(att, dict) and isinstance(att.get("text"), str): + att["text"] = self.__sanitize_text(att["text"]) + + return payload def chat_send_message(self, message): if "rid" in message: From 003d27ecf85126b52145676b42a5b30bb77702a3 Mon Sep 17 00:00:00 2001 From: Leon Buchner Date: Thu, 4 Sep 2025 13:08:30 +0200 Subject: [PATCH 4/5] add chat_test --- tests/test_chat.py | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/test_chat.py b/tests/test_chat.py index 1dc3684..eb39082 100644 --- a/tests/test_chat.py +++ b/tests/test_chat.py @@ -227,3 +227,41 @@ def test_chat_get_mentioned_messages(logged_rocket): assert "messages" in chat_get_mentioned_messages assert len(chat_get_mentioned_messages.get("messages")) > 0 assert chat_get_mentioned_messages.get("messages")[0].get("msg") == "hello @user1" + + +def test_chat_post_sanitizes_text_and_attachments(logged_rocket): + # message text contains double-escaped sequences + msg_text = "hello\\nworld\\t!" + att_text = "line1\\nline2" + + chat_post_message = logged_rocket.chat_post_message( + msg_text, + channel="GENERAL", + attachments=[{"color": "#00ff00", "text": att_text}], + ).json() + + assert chat_post_message.get("success") + mid = chat_post_message["message"]["_id"] + + # retrieve and verify that sequences became real control chars + got = logged_rocket.chat_get_message(msg_id=mid).json() + assert got.get("success") + + posted = got["message"] + assert posted["msg"] == "hello\nworld\t!" + assert posted["attachments"][0]["text"] == "line1\nline2" + + +def test_chat_update_sanitizes_text(logged_rocket): + # create a message first + mid = ( + logged_rocket.chat_post_message("seed", channel="GENERAL") + .json()["message"]["_id"] + ) + # update with escaped content + upd = logged_rocket.chat_update( + room_id="GENERAL", msg_id=mid, text="foo\\nbar" + ).json() + assert upd.get("success") + assert upd["message"]["msg"] == "foo\nbar" + From 33cd2bd76c10c61a7bcc89934ceacea055629797 Mon Sep 17 00:00:00 2001 From: Leon Buchner Date: Thu, 4 Sep 2025 13:18:35 +0200 Subject: [PATCH 5/5] bugfix deepsource --- rocketchat_API/APISections/chat.py | 31 +++++++++++++++++------------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/rocketchat_API/APISections/chat.py b/rocketchat_API/APISections/chat.py index 2d9e9ec..9544f7a 100644 --- a/rocketchat_API/APISections/chat.py +++ b/rocketchat_API/APISections/chat.py @@ -5,19 +5,23 @@ class RocketChatChat(RocketChatBase): def chat_post_message(self, text, room_id=None, channel=None, **kwargs): """Posts a new chat message.""" - if channel: - kwargs["channel"] = channel - if room_id: - kwargs["roomId"] = room_id - if not channel or not room_id: + if not (channel or room_id): raise RocketMissingParamException("roomId or channel required") - if not text: - raise RocketMissingParamException("text required") + + text = "" if text is None else text + payload = {"text": text, **kwargs} - self.__sanitize_payload_text_fields(payload) - return self.call_api_post("chat.postMessage", channel=channel, kwargs=payload) + if channel: + payload["channel"] = channel + if room_id: + payload["roomId"] = room_id + + self._sanitize_payload_text_fields(payload) + + return self.call_api_post("chat.postMessage", kwargs=payload) - def __sanitize_text(self, value: str) -> str: + @staticmethod + def _sanitize_text(value: str) -> str: """Return value with common double-escaped control sequences normalized.""" if not isinstance(value, str): return value @@ -29,7 +33,8 @@ def __sanitize_text(self, value: str) -> str: .replace("\\f", "\f") ) - def __sanitize_payload_text_fields(self, payload: dict) -> dict: + @staticmethod + def _sanitize_payload_text_fields(payload: dict) -> dict: """ In-place sanitize of typical text fields in chat payloads: - payload["text"] @@ -41,13 +46,13 @@ def __sanitize_payload_text_fields(self, payload: dict) -> dict: return payload if "text" in payload and isinstance(payload["text"], str): - payload["text"] = self.__sanitize_text(payload["text"]) + payload["text"] = RocketChatChat._sanitize_text(payload["text"]) attachments = payload.get("attachments") if isinstance(attachments, list): for att in attachments: if isinstance(att, dict) and isinstance(att.get("text"), str): - att["text"] = self.__sanitize_text(att["text"]) + att["text"] = RocketChatChat._sanitize_text(att["text"]) return payload