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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
|
# Copyright 2018 New Vector Ltd
# Copyright 2021 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 List
from matrix_common.regex import glob_to_regex
from synapse.types import JsonDict
from ._base import Config, ConfigError
class RoomDirectoryConfig(Config):
section = "roomdirectory"
def read_config(self, config, **kwargs) -> None:
self.enable_room_list_search = config.get("enable_room_list_search", True)
alias_creation_rules = config.get("alias_creation_rules")
if alias_creation_rules is not None:
self._alias_creation_rules = [
_RoomDirectoryRule("alias_creation_rules", rule)
for rule in alias_creation_rules
]
else:
self._alias_creation_rules = [
_RoomDirectoryRule("alias_creation_rules", {"action": "allow"})
]
room_list_publication_rules = config.get("room_list_publication_rules")
if room_list_publication_rules is not None:
self._room_list_publication_rules = [
_RoomDirectoryRule("room_list_publication_rules", rule)
for rule in room_list_publication_rules
]
else:
self._room_list_publication_rules = [
_RoomDirectoryRule("room_list_publication_rules", {"action": "allow"})
]
def generate_config_section(self, config_dir_path, server_name, **kwargs) -> str:
return """
# Uncomment to disable searching the public room list. When disabled
# blocks searching local and remote room lists for local and remote
# users by always returning an empty list for all queries.
#
#enable_room_list_search: false
# The `alias_creation` option controls who's allowed to create aliases
# on this server.
#
# The format of this option is a list of rules that contain globs that
# match against user_id, room_id and the new alias (fully qualified with
# server name). The action in the first rule that matches is taken,
# which can currently either be "allow" or "deny".
#
# Missing user_id/room_id/alias fields default to "*".
#
# If no rules match the request is denied. An empty list means no one
# can create aliases.
#
# Options for the rules include:
#
# user_id: Matches against the creator of the alias
# alias: Matches against the alias being created
# room_id: Matches against the room ID the alias is being pointed at
# action: Whether to "allow" or "deny" the request if the rule matches
#
# The default is:
#
#alias_creation_rules:
# - user_id: "*"
# alias: "*"
# room_id: "*"
# action: allow
# The `room_list_publication_rules` option controls who can publish and
# which rooms can be published in the public room list.
#
# The format of this option is the same as that for
# `alias_creation_rules`.
#
# If the room has one or more aliases associated with it, only one of
# the aliases needs to match the alias rule. If there are no aliases
# then only rules with `alias: *` match.
#
# If no rules match the request is denied. An empty list means no one
# can publish rooms.
#
# Options for the rules include:
#
# user_id: Matches against the creator of the alias
# room_id: Matches against the room ID being published
# alias: Matches against any current local or canonical aliases
# associated with the room
# action: Whether to "allow" or "deny" the request if the rule matches
#
# The default is:
#
#room_list_publication_rules:
# - user_id: "*"
# alias: "*"
# room_id: "*"
# action: allow
"""
def is_alias_creation_allowed(self, user_id: str, room_id: str, alias: str) -> bool:
"""Checks if the given user is allowed to create the given alias
Args:
user_id: The user to check.
room_id: The room ID for the alias.
alias: The alias being created.
Returns:
True if user is allowed to create the alias
"""
for rule in self._alias_creation_rules:
if rule.matches(user_id, room_id, [alias]):
return rule.action == "allow"
return False
def is_publishing_room_allowed(
self, user_id: str, room_id: str, aliases: List[str]
) -> bool:
"""Checks if the given user is allowed to publish the room
Args:
user_id: The user ID publishing the room.
room_id: The room being published.
aliases: any local aliases associated with the room
Returns:
True if user can publish room
"""
for rule in self._room_list_publication_rules:
if rule.matches(user_id, room_id, aliases):
return rule.action == "allow"
return False
class _RoomDirectoryRule:
"""Helper class to test whether a room directory action is allowed, like
creating an alias or publishing a room.
"""
def __init__(self, option_name: str, rule: JsonDict):
"""
Args:
option_name: Name of the config option this rule belongs to
rule: The rule as specified in the config
"""
action = rule["action"]
user_id = rule.get("user_id", "*")
room_id = rule.get("room_id", "*")
alias = rule.get("alias", "*")
if action in ("allow", "deny"):
self.action = action
else:
raise ConfigError(
"%s rules can only have action of 'allow' or 'deny'" % (option_name,)
)
self._alias_matches_all = alias == "*"
try:
self._user_id_regex = glob_to_regex(user_id)
self._alias_regex = glob_to_regex(alias)
self._room_id_regex = glob_to_regex(room_id)
except Exception as e:
raise ConfigError("Failed to parse glob into regex") from e
def matches(self, user_id: str, room_id: str, aliases: List[str]) -> bool:
"""Tests if this rule matches the given user_id, room_id and aliases.
Args:
user_id: The user ID to check.
room_id: The room ID to check.
aliases: The associated aliases to the room. Will be a single element
for testing alias creation, and can be empty for testing room
publishing.
Returns:
True if the rule matches.
"""
# Note: The regexes are anchored at both ends
if not self._user_id_regex.match(user_id):
return False
if not self._room_id_regex.match(room_id):
return False
# We only have alias checks left, so we can short circuit if the alias
# rule matches everything.
if self._alias_matches_all:
return True
# If we are not given any aliases then this rule only matches if the
# alias glob matches all aliases, which we checked above.
if not aliases:
return False
# Otherwise, we just need one alias to match
for alias in aliases:
if self._alias_regex.match(alias):
return True
return False
|
