yasirarism/pyrofork
1
0
Fork 0
mirror of https://github.com/Mayuri-Chan/pyrofork.git synced 2026-01-09 08:14:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix: Add validation for zero and invalid peer IDs in get_peer_type and get_channel_id - Added validation to raise ValueError for peer_id == 0 in get_peer_type to handle invalid peer IDs. - Improved error handling in get_channel_id by validating peer_id range before conversion. - Ensured descriptive error messages for invalid peer IDs for better debugging.

Browse source
This commit is contained in:
Sahid malik 2024-09-21 23:37:44 +05:30 committed by GitHub
parent 65190eb195
commit bda57d00fa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 28 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
pyrogram/utils.py
View file

@ -282,67 +282,50 @@ MAX_USER_ID = 999999999999
def get_raw_peer_id( def get_raw_peer_id(
peer: Union[ peer: Union[raw.base.Peer, raw.base.RequestedPeer, raw.base.InputPeer]
raw.base.Peer, ) -> Optional[int]:
raw.base.RequestedPeer, """Get the raw peer ID from a Peer object"""
raw.base.InputPeer
] if isinstance(peer, (raw.types.PeerUser, raw.types.RequestedPeerUser, raw.types.InputPeerUser)):
) -> Optional[int]:
"""Get the raw peer id from a Peer object"""
if (
isinstance(peer, raw.types.PeerUser)
or isinstance(peer, raw.types.RequestedPeerUser)
or isinstance(peer, raw.types.InputPeerUser)
):
return peer.user_id return peer.user_id
if ( elif isinstance(peer, (raw.types.PeerChat, raw.types.RequestedPeerChat, raw.types.InputPeerChat)):
isinstance(peer, raw.types.PeerChat)
or isinstance(peer, raw.types.RequestedPeerChat)
or isinstance(peer, raw.types.InputPeerChat)
):
return peer.chat_id return peer.chat_id
if ( elif isinstance(peer, (raw.types.PeerChannel, raw.types.RequestedPeerChannel, raw.types.InputPeerChannel)):
isinstance(peer, raw.types.PeerChannel)
or isinstance(peer, raw.types.RequestedPeerChannel)
or isinstance(peer, raw.types.InputPeerChannel)
):
return peer.channel_id return peer.channel_id
return None return None # Return None explicitly when peer type does not match
def get_peer_id(peer: Union[raw.base.Peer, raw.base.InputPeer]) -> int: def get_peer_id(peer: Union[raw.base.Peer, raw.base.InputPeer]) -> int:
"""Get the non-raw peer id from a Peer object""" """Get the non-raw peer ID from a Peer object"""
if (
isinstance(peer, raw.types.PeerUser) if isinstance(peer, (raw.types.PeerUser, raw.types.InputPeerUser)):
or isinstance(peer, raw.types.InputPeerUser)
):
return peer.user_id return peer.user_id
if ( elif isinstance(peer, (raw.types.PeerChat, raw.types.InputPeerChat)):
isinstance(peer, raw.types.PeerChat) return -peer.chat_id # Return negative chat_id for PeerChat
or isinstance(peer, raw.types.InputPeerChat)
):
return -peer.chat_id
if ( elif isinstance(peer, (raw.types.PeerChannel, raw.types.InputPeerChannel)):
isinstance(peer, raw.types.PeerChannel) return MAX_CHANNEL_ID - peer.channel_id # Return ID based on MAX_CHANNEL_ID for PeerChannel
or isinstance(peer, raw.types.InputPeerChannel)
):
return MAX_CHANNEL_ID - peer.channel_id
raise ValueError(f"Peer type invalid: {peer}")
# Raise an error if an unknown peer type is encountered
raise ValueError(f"Invalid Peer type: {type(peer).__name__}")
def get_peer_type(peer_id: int) -> str: def get_peer_type(peer_id: int) -> str:
"""Determine the type of peer from the peer_id."""
if peer_id == 0:
raise ValueError("Peer id cannot be zero")
if peer_id < 0: if peer_id < 0:
if MIN_CHAT_ID <= peer_id: if MIN_CHAT_ID <= peer_id:
return "chat" return "chat"
if MIN_CHANNEL_ID <= peer_id < MAX_CHANNEL_ID: if MIN_CHANNEL_ID <= peer_id < MAX_CHANNEL_ID:
return "channel" return "channel"
elif 0 < peer_id <= MAX_USER_ID: elif 0 < peer_id <= MAX_USER_ID:
return "user" return "user"
@ -350,8 +333,12 @@ def get_peer_type(peer_id: int) -> str:
def get_channel_id(peer_id: int) -> int: def get_channel_id(peer_id: int) -> int:
return MAX_CHANNEL_ID - peer_id """Convert peer_id to a channel id."""
# Check if it's a valid channel peer ID before conversion
if not (MIN_CHANNEL_ID <= peer_id < MAX_CHANNEL_ID):
raise ValueError(f"Invalid channel peer id: {peer_id}")
return MAX_CHANNEL_ID - peer_id
def btoi(b: bytes) -> int: def btoi(b: bytes) -> int:
return int.from_bytes(b, "big") return int.from_bytes(b, "big")