Skip to content

fix: Admins promoting logic fixed #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danoctua merged 1 commit into from
Apr 17, 2025
Merged

fix: Admins promoting logic fixed #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 96 additions & 68 deletions backend/community_manager/handlers/chat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from core.services.chat import TelegramChatService
from core.services.chat.user import TelegramChatUserService
from core.services.db import DBService
from core.services.user import UserService
from core.utils.events import ChatJoinRequestEventBuilder, ChatAdminChangeEventBuilder

logger = logging.getLogger(__name__)
Expand All @@ -27,7 +28,7 @@ async def handle_chat_action(event: events.ChatAction.Event):
telegram_chat_service = TelegramChatService(session)
if not telegram_chat_service.check_exists(event.chat_id):
logger.debug(
"Chat doesn't exist, but bot was not added to the chat: %d. Skipping",
"Chat doesn't exist, but bot was not added to the chat: %d. Skipping event...",
)
return

Expand All @@ -40,7 +41,20 @@ async def handle_chat_action(event: events.ChatAction.Event):
telegram_chat_action = TelegramChatAction(
session, telethon_client=event.client
)
await telegram_chat_action.fetch_and_push_profile_photo(event.chat)
logo_path = await telegram_chat_action.fetch_and_push_profile_photo(
event.chat
)
if logo_path:
logger.debug(
f"Updating logo for chat {event.chat_id!r} with path {logo_path.name!r}..."
)
telegram_chat_service.set_logo(
chat_id=event.chat_id, logo_path=logo_path.name
)
else:
logger.warning(
f"Ignoring update for chat {event.chat_id!r} as logo was not downloaded.."
)
return

logger.debug(
Expand All @@ -58,13 +72,11 @@ async def handle_chat_action(event: events.ChatAction.Event):
elif event.action_message:
# To avoid handling multiple events twice (users are added or removed and the action message is sent)
# The only message that should be handled is when bot is added to the chat
logger.debug(
f"Chat action message {event!r} is not handled, but was removed."
)
logger.debug(f"Chat action message {event!r} is not handled.")
return

elif event.user.bot and not event.user.is_self:
logger.debug(f"Other bot user {event.user.id!r} is not handled.")
logger.debug(f"Another bot user {event.user.id!r} is not handled.")
return

authorization_action = AuthorizationAction(
Expand All @@ -86,7 +98,7 @@ async def handle_chat_action(event: events.ChatAction.Event):
return

logger.info(
f"New chat member: {event.chat_id=!r} {event.user=!r}",
f"New chat member: {event.chat_id=!r} {event.user.id=!r}",
extra={"event": event},
)
await authorization_action.on_chat_member_in(
Expand All @@ -109,7 +121,7 @@ async def handle_chat_action(event: events.ChatAction.Event):
return

logger.info(
f"Chat member left/kicked: {event.chat_id=!r} {event.user=!r}",
f"Chat member left/kicked: {event.chat_id=!r} {event.user_id=!r}",
extra={"event": event},
)
await authorization_action.on_chat_member_out(
Expand All @@ -118,7 +130,7 @@ async def handle_chat_action(event: events.ChatAction.Event):
)

else:
logger.info(f"Unhandled chat action: {event!r}")
logger.debug(f"Unhandled chat action: {event!r}")


async def handle_join_request(event: ChatJoinRequestEventBuilder.Event):
Expand Down Expand Up @@ -149,87 +161,103 @@ async def handle_chat_participant_update(
)
with DBService().db_session() as session:
telegram_chat_service = TelegramChatService(session)
chat: TelegramChatDTO
chat: TelegramChatDTO | None = None
try:
chat = TelegramChatDTO.from_object(telegram_chat_service.get(chat_id))
# Handle bot updates only if chat existed previously
if event.is_self:
logger.info(
"Bot user %d is managed in the chat %d: %s",
event.user.id,
chat_id,
event.new_participant,
)
if not event.sufficient_bot_privileges:
if not chat.insufficient_privileges:
logger.warning(
"Insufficient permissions for the bot in chat %d", chat_id
)
telegram_chat_service.set_insufficient_privileges(
chat_id=chat_id, value=True
)
else:
if chat.insufficient_privileges:
logger.info(
"Sufficient permissions for the bot in chat %d", chat_id
)
telegram_chat_service.set_insufficient_privileges(
chat_id=chat_id, value=False
)
return

except NoResultFound:
# Otherwise create a new chat if bot privileges are sufficient for this
logger.info(
f"Chat {chat_id!r} doesn't exist, but bot was added with admin rights. Creating..."
)
pass

if not chat:
# If bot is able to see that update, it means it was promoted to admins
# at some point, so the chat should be created in the database
if event.sufficient_bot_privileges:
if event.is_self and event.sufficient_bot_privileges:
# Create a new chat if bot privileges are sufficient for this
logger.info(
f"Chat {chat_id!r} doesn't exist, but bot was added with admin rights. Creating..."
)
telegram_chat_action = TelegramChatAction(
session, telethon_client=event.client
)
await telegram_chat_action.create(
chat_id,
sufficient_bot_privileges=event.sufficient_bot_privileges,
)
else:
logger.debug(
f"Chat {chat_id!r} doesn't exist, but bot was added without admin rights. Skipping."
)
return

# If chat already existed
logger.debug("Handling chat participant update %s", event)

if (target_user_entity := event.user) and target_user_entity.bot:
logger.debug(f"Bot user {target_user_entity.id!r} is not handled.")
# Handling updates for the bot user
if event.is_self:
logger.info(
"The Bot user is managed in the chat %d: %s",
chat_id,
event.new_participant,
)
if not event.sufficient_bot_privileges:
if not chat.insufficient_privileges:
logger.warning(
"Insufficient permissions for the bot in chat %d", chat_id
)
telegram_chat_service.set_insufficient_privileges(
chat_id=chat_id, value=True
)
else:
if chat.insufficient_privileges:
logger.info(
"Sufficient permissions for the bot in chat %d", chat_id
)
telegram_chat_service.set_insufficient_privileges(
chat_id=chat_id, value=False
)
return

# Ignore actions done on other bots
if (target_telethon_user := event.user) and target_telethon_user.bot:
logger.debug(f"Bot user {target_telethon_user.id!r} is not handled.")
return

user_service = UserService(session)
target_user = user_service.get_or_create(
telegram_user=TelegramUserDTO.from_telethon_user(target_telethon_user)
)
telegram_chat_user_service = TelegramChatUserService(db_session=session)

try:
target_user = telegram_chat_user_service.get(
chat_id=chat_id, user_id=target_user_entity.id
target_chat_user = telegram_chat_user_service.get(
chat_id=chat.id, user_id=target_user.id
)

if event.is_demoted:
logger.info(
"Admin %d demoted in chat %d", target_user_entity.id, chat_id
)
if target_user.is_admin:
telegram_chat_user_service.demote_admin(
chat_id=chat_id, user_id=target_user.user_id
)
return

elif event.is_promoted:
logger.info(
"Admin %d promoted in chat %d", target_user_entity.id, chat_id
)
if not target_user.is_admin:
telegram_chat_user_service.promote_admin(
chat_id=chat_id, user_id=target_user.user_id
)
return
except NoResultFound:
logger.info(
f"No chat user found in chat {chat.id!r} for user {target_user.id!r}. Creating..."
)
telegram_chat_user_service.create(
chat_id=chat_id,
user_id=target_user_entity.id,
is_admin=bool(getattr(event.new_participant, "admin_rights", None)),
chat_id=chat.id,
user_id=target_user.id,
is_admin=event.has_enough_rights,
# Because it was not added by the bot user
is_managed=False,
)
return

# Handle admin privileges update on the normal user
if event.is_demoted or not event.has_enough_rights:
if target_chat_user.is_admin:
logger.info("Admin %d demoted in chat %d", target_user.id, chat.id)
telegram_chat_user_service.demote_admin(
chat_id=chat.id, user_id=target_chat_user.user_id
)
return

elif event.has_enough_rights:
if not target_chat_user.is_admin:
logger.info("Admin %d promoted in chat %d", target_user.id, chat.id)
telegram_chat_user_service.promote_admin(
chat_id=chat.id,
user_id=target_user.id,
)
return
3 changes: 3 additions & 0 deletions backend/core/constants.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,9 @@

ASYNC_TASK_REDIS_PREFIX = "atask"

# Privileges required for admin to manage the chat in the bot
REQUIRED_ADMIN_PRIVILEGES = ["add_admins"]
# Privileges required for a bot user to manage the chat
REQUIRED_BOT_PRIVILEGES = ["invite_users", "ban_users"]

# ------------------ Redis --------------------
Expand Down
15 changes: 15 additions & 0 deletions backend/core/services/chat/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,21 @@ def check_exists(self, chat_id: int) -> bool:
> 0
)

def set_logo(self, chat_id: int, logo_path: str) -> None:
"""
Updates the logo path for a specified Telegram chat in the database by its
chat ID. Commits the changes to the database and logs the operation.

:param chat_id: The unique identifier of the Telegram chat.
:param logo_path: The file path of the logo to be set.
:return: None
"""
self.db_session.query(TelegramChat).filter(TelegramChat.id == chat_id).update(
{"logo_path": logo_path}
)
self.db_session.commit()
logger.debug(f"Telegram Chat {chat_id!r} logo set.")

def clear_logo(self, chat_id: int) -> None:
chat = self.get(chat_id)
chat.logo_path = None
Expand Down
36 changes: 33 additions & 3 deletions backend/core/utils/events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,7 @@
ChannelParticipantBanned,
)

from core.constants import REQUIRED_BOT_PRIVILEGES

from core.constants import REQUIRED_BOT_PRIVILEGES, REQUIRED_ADMIN_PRIVILEGES

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -112,9 +111,40 @@ def is_promoted(self) -> bool:
self.prev_participant, ChannelParticipantAdmin
) and isinstance(self.new_participant, ChannelParticipantAdmin)

@property
def has_enough_rights(self) -> bool:
if self.user.bot:
logger.warning(
f"This method should not be used on the bot users: {self.original_update}"
)
return False

if not isinstance(self.new_participant, ChannelParticipantAdmin):
logger.debug(
"User %d is not an admin in the chat %d",
self.user.id,
self.original_update.channel_id,
)
return False
elif not all(
[
getattr(self.new_participant.admin_rights, right)
for right in REQUIRED_ADMIN_PRIVILEGES
]
):
logger.warning(
"User %d has insufficient permissions in the chat %d: %s",
self.user.id,
self.original_update.channel_id,
self.new_participant.admin_rights,
)
return False

return True

@property
def sufficient_bot_privileges(self) -> bool:
if not self.is_self:
if not self.new_participant.is_self:
raise ValueError("The event is not related to the bot user.")

if not isinstance(self.new_participant, ChannelParticipantAdmin):
Expand Down
Loading