feat: use lib side check of valid lengths and prevent invalid requests being sent.

pyrogram/client.py

@@ -362,6 +362,8 @@ def __init__(
         self.message_cache = Cache(self.max_message_cache_size)
         self.business_user_connection_cache = Cache(self.max_business_user_connection_cache_size)
 
+        self.app_constant = Constant(client_instance=self)
+
         # Sometimes, for some reason, the server will stop sending updates and will only respond to pings.
         # This watchdog will invoke updates.GetState in order to wake up the server and enable it sending updates again
         # after some idle time has been detected.
@@ -1242,3 +1244,47 @@ def __setitem__(self, key, value):
         if len(self.store) > self.capacity:
             for _ in range(self.capacity // 2 + 1):
                 del self.store[next(iter(self.store))]
+
+
+class Constant:
+    """Tuple of min, max, premium max lengths"""
+
+    # Text of the message to be sent, 1-4096 characters
+    MAX_TEXT_LENGTH = (1, 4096, 0)
+
+    # Caption for the animation, audio, document, photo, video or voice, 0-1024 characters
+    MAX_CAPTION_LENGTH = (0, 1024, 0)
+
+    MAX_FIRST_NAME_LENGTH = (1, 64, 0)
+    MAX_LAST_NAME_LENGTH = (0, 64, 0)
+
+    # USER's BIO
+    MAX_BIO_LENGTH = (0, 70, 140)
+
+    # ADMIN TITLE
+    MAX_TITLE_LENGTH = (0, 16, 0)
+
+    # Use the InlineQuery.answer() method. No more than 50 results per query are allowed.
+    MAX_RESULTS_LENGTH = (0, 50, 0)
+
+    def __init__(self, client_instance: Client):
+        self.is_premium = client_instance.me.is_premium
+
+    def check_valid_length(self, text: Union[List, str], arg_type: str, ):
+        if not isinstance(text, (str, list)):
+            raise ValueError(f"Argument {arg_type} must be a str | list")
+
+        text_length = len(text)
+
+        attr_str = f"max_{arg_type}_length".upper()
+
+        # MIN, MAX, PREM-MAX
+        lengths: tuple[int, int, int] = getattr(self, attr_str, (0, 0, 0))
+
+        min_length = lengths[0]
+        # Check if client is premium and the premium value isn't 0
+        max_length = lengths[2] if self.is_premium and lengths[2] != 0 else lengths[1]
+
+        error_info = f"\nInvalid length of {text_length} for arg {arg_type}\nValid Lengths: {min_length}-{max_length}"
+
+        assert bool(min_length < text_length <= max_length), error_info

pyrogram/methods/bots/answer_inline_query.py

@@ -96,6 +96,8 @@ async def answer_inline_query(
                             InputTextMessageContent("Message content"))])
         """
 
+        self.app_constant.check_valid_length(text=results, arg_type="results")
+
         return await self.invoke(
             raw.functions.messages.SetInlineBotResults(
                 query_id=int(inline_query_id),

pyrogram/methods/chats/set_administrator_title.py

@@ -16,7 +16,7 @@
 #  You should have received a copy of the GNU Lesser General Public License
 #  along with Pyrogram.  If not, see <http://www.gnu.org/licenses/>.
 
-from typing import Union
+from typing import Optional, Union
 
 import pyrogram
 from pyrogram import raw
@@ -27,7 +27,7 @@ async def set_administrator_title(
         self: "pyrogram.Client",
         chat_id: Union[int, str],
         user_id: Union[int, str],
-        title: str,
+        title: Optional[str] = None,
     ) -> bool:
         """Set a custom title (rank) to an administrator of a supergroup.
 
@@ -45,6 +45,7 @@ async def set_administrator_title(
                 For a contact that exists in your Telegram address book you can use his phone number (str).
 
             title (``str``, *optional*):
+                New custom title for the administrator; 0-16 characters, emoji are not allowed.
                 A custom title that will be shown to all members instead of "Owner" or "Admin".
                 Pass None or "" (empty string) to remove the custom title.
 
@@ -56,6 +57,10 @@ async def set_administrator_title(
 
                 await app.set_administrator_title(chat_id, user_id, "Admin Title")
         """
+
+        if title:
+            self.app_constant.check_valid_length(text=title, arg_type="title")
+
         chat_id = await self.resolve_peer(chat_id)
         user_id = await self.resolve_peer(user_id)
 

pyrogram/methods/messages/edit_message_text.py

@@ -94,6 +94,7 @@ async def edit_message_text(
                     )
                 )
         """
+
         if disable_web_page_preview and link_preview_options:
             raise ValueError(
                 "Parameters `disable_web_page_preview` and `link_preview_options` are mutually "

pyrogram/methods/messages/send_message.py

@@ -157,6 +157,7 @@ async def send_message(
                             ]
                         ]))
         """
+
         if disable_web_page_preview and link_preview_options:
             raise ValueError(
                 "Parameters `disable_web_page_preview` and `link_preview_options` are mutually "
@@ -192,6 +193,9 @@ async def send_message(
         )
         message, entities = (await utils.parse_text_entities(self, text, parse_mode, entities)).values()
 
+        if message:
+            self.app_constant.check_valid_length(text=message, arg_type="text")
+
         session = None
         business_connection = None
         if business_connection_id:

pyrogram/methods/users/update_profile.py

@@ -63,6 +63,10 @@ async def update_profile(
                 await app.update_profile(last_name="")
         """
 
+        for var_name in ("first_name", "last_name", "bio"):
+            if var_value := locals()[var_name]:
+                self.app_constant.check_valid_length(text=var_value, arg_type=var_name)
+
         return bool(
             await self.invoke(
                 raw.functions.account.UpdateProfile(

pyrogram/session/session.py

@@ -125,7 +125,7 @@ async def start(self):
                 await self.send(raw.functions.Ping(ping_id=0), timeout=self.START_TIMEOUT)
 
                 if not self.is_cdn:
-                    await self.send(
+                    cfg = await self.send(
                         raw.functions.InvokeWithLayer(
                             layer=layer,
                             query=raw.functions.InitConnection(
@@ -146,6 +146,9 @@ async def start(self):
                         ),
                         timeout=self.START_TIMEOUT
                     )
+                    if isinstance(cfg, raw.types.Config):  # TODO
+                        self.client.app_constant.MAX_TEXT_LENGTH = (1, cfg.message_length_max, cfg.message_length_max)
+                        self.client.app_constant.MAX_CAPTION_LENGTH = (0, cfg.caption_length_max, cfg.caption_length_max)
 
                 self.ping_task = self.loop.create_task(self.ping_worker())