You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
114 lines
2.8 KiB
Python
114 lines
2.8 KiB
Python
from __future__ import annotations
|
|
"""
|
|
InvitePolicy class:
|
|
InvitePolicy object are use to chose whether to accept or decline
|
|
an invite to a room.
|
|
"""
|
|
|
|
import asyncio
|
|
import nio
|
|
import logging
|
|
|
|
from abc import (
|
|
ABC,
|
|
abstractmethod
|
|
)
|
|
from typing import (
|
|
Union,
|
|
TYPE_CHECKING
|
|
)
|
|
if TYPE_CHECKING:
|
|
from .client import Client
|
|
from .utils import (
|
|
Room,
|
|
RoomAlias,
|
|
RoomId
|
|
)
|
|
from .async_utils import Aobject
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
class InvitePolicy(ABC):
|
|
"""
|
|
Class used by Client to chose whether to accept of decline an invite.
|
|
"""
|
|
|
|
@abstractmethod
|
|
async def accept_invite(
|
|
self,
|
|
room_id: RoomId,
|
|
invite: nio.responses.InviteInfo
|
|
)->bool:
|
|
"""
|
|
Test if the invit must be accepted of declined.
|
|
Async because the policy might want to do exotic
|
|
stuff, like, idk, send a pm to someone to ask confirmation
|
|
"""
|
|
pass
|
|
|
|
class DeclineAll(InvitePolicy):
|
|
"""
|
|
Decline all invitations.
|
|
"""
|
|
|
|
async def accept_invite(
|
|
self,
|
|
room_id: RoomId,
|
|
invite: nio.responses.InviteInfo
|
|
)->bool:
|
|
return False
|
|
|
|
class AcceptAll(InvitePolicy):
|
|
"""
|
|
Accept all invitations.
|
|
"""
|
|
|
|
async def accept_invite(
|
|
self,
|
|
room_id: RoomId,
|
|
invite: nio.responses.InviteInfo
|
|
)->bool:
|
|
return True
|
|
|
|
class WhiteList(InvitePolicy, Aobject):
|
|
"""
|
|
Accept invite for whitelisted room.
|
|
This policy cannot be set during the initialization of the
|
|
client because we need the client to initialize this object:
|
|
use `client.set_invite_policy(policy)`
|
|
"""
|
|
|
|
whitelisted_rooms: dict[RoomId, RoomAlias]
|
|
|
|
async def __init__(
|
|
self,
|
|
client: Client,
|
|
whitelisted_rooms_names: list[Union[RoomAlias, RoomId]]
|
|
):
|
|
"""
|
|
client: the matrix client (cf client.py)
|
|
whitelisted_rooms: the list of the rooms where the bot is allowed to connect
|
|
(given by room id (expl: '!xxx:matrix.org') or room alias (expl:
|
|
'#xxx:matrix.org'))
|
|
"""
|
|
self.whitelisted_rooms = {}
|
|
rooms = await asyncio.gather(*(client.resolve_room(room_name) for room_name in whitelisted_rooms_names))
|
|
for room in rooms:
|
|
self.whitelisted_rooms[room.id] = room
|
|
|
|
async def accept_invite(
|
|
self,
|
|
room_id: RoomId,
|
|
invite: nio.responses.InviteInfo
|
|
)->bool:
|
|
"""
|
|
Accept invite from whitelisted room, reject the other.
|
|
"""
|
|
if room_id in self.whitelisted_rooms:
|
|
log.info(f"Invite from {room_id} accepted.")
|
|
return True
|
|
else:
|
|
log.warning(f"Invite from {room_id} declined: not in whitelist.")
|
|
return False
|
|
|