summary refs log tree commit diff
path: root/synapse/events/third_party_rules.py
blob: 1ca77519d59afc17bd2208f2b1bc1f6a6469f100 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
# -*- coding: utf-8 -*-
# Copyright 2019 The Matrix.org Foundation C.I.C.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import Callable

from synapse.events import EventBase
from synapse.events.snapshot import EventContext
from synapse.module_api import ModuleApi
from synapse.types import Requester, StateMap


class ThirdPartyEventRules:
    """Allows server admins to provide a Python module implementing an extra
    set of rules to apply when processing events.

    This is designed to help admins of closed federations with enforcing custom
    behaviours.
    """

    def __init__(self, hs):
        self.third_party_rules = None

        self.store = hs.get_datastore()

        module = None
        config = None
        if hs.config.third_party_event_rules:
            module, config = hs.config.third_party_event_rules

        if module is not None:
            self.third_party_rules = module(
                config=config, module_api=ModuleApi(hs, hs.get_auth_handler()),
            )

    async def check_event_allowed(
        self, event: EventBase, context: EventContext
    ) -> bool:
        """Check if a provided event should be allowed in the given context.

        Args:
            event: The event to be checked.
            context: The context of the event.

        Returns:
            True if the event should be allowed, False if not.
        """
        if self.third_party_rules is None:
            return True

        prev_state_ids = await context.get_prev_state_ids()

        # Retrieve the state events from the database.
        state_events = {}
        for key, event_id in prev_state_ids.items():
            state_events[key] = await self.store.get_event(event_id, allow_none=True)

        ret = await self.third_party_rules.check_event_allowed(event, state_events)
        return ret

    async def on_create_room(
        self, requester: Requester, config: dict, is_requester_admin: bool
    ) -> bool:
        """Intercept requests to create room to allow, deny or update the
        request config.

        Args:
            requester
            config: The creation config from the client.
            is_requester_admin: If the requester is an admin

        Returns:
            Whether room creation is allowed or denied.
        """

        if self.third_party_rules is None:
            return True

        ret = await self.third_party_rules.on_create_room(
            requester, config, is_requester_admin
        )
        return ret

    async def check_threepid_can_be_invited(
        self, medium: str, address: str, room_id: str
    ) -> bool:
        """Check if a provided 3PID can be invited in the given room.

        Args:
            medium: The 3PID's medium.
            address: The 3PID's address.
            room_id: The room we want to invite the threepid to.

        Returns:
            True if the 3PID can be invited, False if not.
        """

        if self.third_party_rules is None:
            return True

        state_events = await self._get_state_map_for_room(room_id)

        ret = await self.third_party_rules.check_threepid_can_be_invited(
            medium, address, state_events
        )
        return ret

    async def check_visibility_can_be_modified(
        self, room_id: str, new_visibility: str
    ) -> bool:
        """Check if a room is allowed to be published to, or removed from, the public room
        list.

        Args:
            room_id: The ID of the room.
            new_visibility: The new visibility state. Either "public" or "private".

        Returns:
            True if the room's visibility can be modified, False if not.
        """
        if self.third_party_rules is None:
            return True

        check_func = getattr(
            self.third_party_rules, "check_visibility_can_be_modified", None
        )
        if not check_func or not isinstance(check_func, Callable):
            return True

        state_events = await self._get_state_map_for_room(room_id)

        return await check_func(room_id, state_events, new_visibility)

    async def _get_state_map_for_room(self, room_id: str) -> StateMap[EventBase]:
        """Given a room ID, return the state events of that room.

        Args:
            room_id: The ID of the room.

        Returns:
            A dict mapping (event type, state key) to state event.
        """
        state_ids = await self.store.get_filtered_current_state_ids(room_id)
        room_state_events = await self.store.get_events(state_ids.values())

        state_events = {}
        for key, event_id in state_ids.items():
            state_events[key] = room_state_events[event_id]

        return state_events