Merge branch 'develop' of github.com:matrix-org/synapse into matthew/free_mau_alt

This commit is contained in:
Erik Johnston 2018-08-23 18:10:32 +01:00
commit 42620b4a92
16 changed files with 484 additions and 34 deletions

1
changelog.d/3680.feature Normal file
View File

@ -0,0 +1 @@
Server notices for resource limit blocking

1
changelog.d/3734.misc Normal file
View File

@ -0,0 +1 @@
Reference the need for an HTTP replication port when using the federation_reader worker

1
changelog.d/3746.misc Normal file
View File

@ -0,0 +1 @@
Fix MAU cache invalidation due to missing yield

1
changelog.d/3747.bugfix Normal file
View File

@ -0,0 +1 @@
Fix bug where we resent "limit exceeded" server notices repeatedly

View File

@ -74,7 +74,7 @@ replication endpoints that it's talking to on the main synapse process.
``worker_replication_port`` should point to the TCP replication listener port and
``worker_replication_http_port`` should point to the HTTP replication port.
Currently, only the ``event_creator`` worker requires specifying
Currently, the ``event_creator`` and ``federation_reader`` workers require specifying
``worker_replication_http_port``.
For instance::

View File

@ -783,10 +783,16 @@ class Auth(object):
user_id(str|None): If present, checks for presence against existing
MAU cohort
"""
# Never fail an auth check for the server notices users
# This can be a problem where event creation is prohibited due to blocking
if user_id == self.hs.config.server_notices_mxid:
return
if self.hs.config.hs_disabled:
raise ResourceLimitError(
403, self.hs.config.hs_disabled_message,
errcode=Codes.RESOURCE_LIMIT_EXCEED,
errcode=Codes.RESOURCE_LIMIT_EXCEEDED,
admin_uri=self.hs.config.admin_uri,
limit_type=self.hs.config.hs_disabled_limit_type
)
@ -803,6 +809,6 @@ class Auth(object):
403, "Monthly Active User Limit Exceeded",
admin_uri=self.hs.config.admin_uri,
errcode=Codes.RESOURCE_LIMIT_EXCEED,
errcode=Codes.RESOURCE_LIMIT_EXCEEDED,
limit_type="monthly_active_user"
)

View File

@ -78,6 +78,7 @@ class EventTypes(object):
Name = "m.room.name"
ServerACL = "m.room.server_acl"
Pinned = "m.room.pinned_events"
class RejectedReason(object):
@ -108,3 +109,6 @@ DEFAULT_ROOM_VERSION = RoomVersions.V1
# vdh-test-version is a placeholder to get room versioning support working and tested
# until we have a working v2.
KNOWN_ROOM_VERSIONS = {RoomVersions.V1, RoomVersions.VDH_TEST}
ServerNoticeMsgType = "m.server_notice"
ServerNoticeLimitReached = "m.server_notice.usage_limit_reached"

View File

@ -56,7 +56,7 @@ class Codes(object):
SERVER_NOT_TRUSTED = "M_SERVER_NOT_TRUSTED"
CONSENT_NOT_GIVEN = "M_CONSENT_NOT_GIVEN"
CANNOT_LEAVE_SERVER_NOTICE_ROOM = "M_CANNOT_LEAVE_SERVER_NOTICE_ROOM"
RESOURCE_LIMIT_EXCEED = "M_RESOURCE_LIMIT_EXCEED"
RESOURCE_LIMIT_EXCEEDED = "M_RESOURCE_LIMIT_EXCEEDED"
UNSUPPORTED_ROOM_VERSION = "M_UNSUPPORTED_ROOM_VERSION"
INCOMPATIBLE_ROOM_VERSION = "M_INCOMPATIBLE_ROOM_VERSION"
@ -238,7 +238,7 @@ class ResourceLimitError(SynapseError):
"""
def __init__(
self, code, msg,
errcode=Codes.RESOURCE_LIMIT_EXCEED,
errcode=Codes.RESOURCE_LIMIT_EXCEEDED,
admin_uri=None,
limit_type=None,
):

View File

@ -0,0 +1,195 @@
# -*- coding: utf-8 -*-
# Copyright 2018 New Vector Ltd
#
# 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.
import logging
from six import iteritems
from twisted.internet import defer
from synapse.api.constants import (
EventTypes,
ServerNoticeLimitReached,
ServerNoticeMsgType,
)
from synapse.api.errors import AuthError, ResourceLimitError, SynapseError
from synapse.server_notices.server_notices_manager import SERVER_NOTICE_ROOM_TAG
logger = logging.getLogger(__name__)
class ResourceLimitsServerNotices(object):
""" Keeps track of whether the server has reached it's resource limit and
ensures that the client is kept up to date.
"""
def __init__(self, hs):
"""
Args:
hs (synapse.server.HomeServer):
"""
self._server_notices_manager = hs.get_server_notices_manager()
self._store = hs.get_datastore()
self._auth = hs.get_auth()
self._config = hs.config
self._resouce_limited = False
self._message_handler = hs.get_message_handler()
self._state = hs.get_state_handler()
@defer.inlineCallbacks
def maybe_send_server_notice_to_user(self, user_id):
"""Check if we need to send a notice to this user, this will be true in
two cases.
1. The server has reached its limit does not reflect this
2. The room state indicates that the server has reached its limit when
actually the server is fine
Args:
user_id (str): user to check
Returns:
Deferred
"""
if self._config.hs_disabled is True:
return
if self._config.limit_usage_by_mau is False:
return
timestamp = yield self._store.user_last_seen_monthly_active(user_id)
if timestamp is None:
# This user will be blocked from receiving the notice anyway.
# In practice, not sure we can ever get here
return
# Determine current state of room
room_id = yield self._server_notices_manager.get_notice_room_for_user(user_id)
if not room_id:
logger.warn("Failed to get server notices room")
return
yield self._check_and_set_tags(user_id, room_id)
currently_blocked, ref_events = yield self._is_room_currently_blocked(room_id)
try:
# Normally should always pass in user_id if you have it, but in
# this case are checking what would happen to other users if they
# were to arrive.
try:
yield self._auth.check_auth_blocking()
is_auth_blocking = False
except ResourceLimitError as e:
is_auth_blocking = True
event_content = e.msg
event_limit_type = e.limit_type
if currently_blocked and not is_auth_blocking:
# Room is notifying of a block, when it ought not to be.
# Remove block notification
content = {
"pinned": ref_events
}
yield self._server_notices_manager.send_notice(
user_id, content, EventTypes.Pinned, '',
)
elif not currently_blocked and is_auth_blocking:
# Room is not notifying of a block, when it ought to be.
# Add block notification
content = {
'body': event_content,
'msgtype': ServerNoticeMsgType,
'server_notice_type': ServerNoticeLimitReached,
'admin_uri': self._config.admin_uri,
'limit_type': event_limit_type
}
event = yield self._server_notices_manager.send_notice(
user_id, content, EventTypes.Message,
)
content = {
"pinned": [
event.event_id,
]
}
yield self._server_notices_manager.send_notice(
user_id, content, EventTypes.Pinned, '',
)
except SynapseError as e:
logger.error("Error sending resource limits server notice: %s", e)
@defer.inlineCallbacks
def _check_and_set_tags(self, user_id, room_id):
"""
Since server notices rooms were originally not with tags,
important to check that tags have been set correctly
Args:
user_id(str): the user in question
room_id(str): the server notices room for that user
"""
tags = yield self._store.get_tags_for_user(user_id)
server_notices_tags = tags.get(room_id)
need_to_set_tag = True
if server_notices_tags:
if server_notices_tags.get(SERVER_NOTICE_ROOM_TAG):
# tag already present, nothing to do here
need_to_set_tag = False
if need_to_set_tag:
yield self._store.add_tag_to_room(
user_id, room_id, SERVER_NOTICE_ROOM_TAG, None
)
@defer.inlineCallbacks
def _is_room_currently_blocked(self, room_id):
"""
Determines if the room is currently blocked
Args:
room_id(str): The room id of the server notices room
Returns:
bool: Is the room currently blocked
list: The list of pinned events that are unrelated to limit blocking
This list can be used as a convenience in the case where the block
is to be lifted and the remaining pinned event references need to be
preserved
"""
currently_blocked = False
pinned_state_event = None
try:
pinned_state_event = yield self._state.get_current_state(
room_id, event_type=EventTypes.Pinned
)
except AuthError:
# The user has yet to join the server notices room
pass
referenced_events = []
if pinned_state_event is not None:
referenced_events = list(pinned_state_event.content.get('pinned', []))
events = yield self._store.get_events(referenced_events)
for event_id, event in iteritems(events):
if event.type != EventTypes.Message:
continue
if event.content.get("msgtype") == ServerNoticeMsgType:
currently_blocked = True
# remove event in case we need to disable blocking later on.
if event_id in referenced_events:
referenced_events.remove(event.event_id)
defer.returnValue((currently_blocked, referenced_events))

View File

@ -22,6 +22,8 @@ from synapse.util.caches.descriptors import cachedInlineCallbacks
logger = logging.getLogger(__name__)
SERVER_NOTICE_ROOM_TAG = "m.server_notice"
class ServerNoticesManager(object):
def __init__(self, hs):
@ -46,7 +48,10 @@ class ServerNoticesManager(object):
return self._config.server_notices_mxid is not None
@defer.inlineCallbacks
def send_notice(self, user_id, event_content):
def send_notice(
self, user_id, event_content,
type=EventTypes.Message, state_key=None
):
"""Send a notice to the given user
Creates the server notices room, if none exists.
@ -54,9 +59,11 @@ class ServerNoticesManager(object):
Args:
user_id (str): mxid of user to send event to.
event_content (dict): content of event to send
type(EventTypes): type of event
is_state_event(bool): Is the event a state event
Returns:
Deferred[None]
Deferred[FrozenEvent]
"""
room_id = yield self.get_notice_room_for_user(user_id)
@ -65,15 +72,20 @@ class ServerNoticesManager(object):
logger.info("Sending server notice to %s", user_id)
yield self._event_creation_handler.create_and_send_nonmember_event(
requester, {
"type": EventTypes.Message,
"room_id": room_id,
"sender": system_mxid,
"content": event_content,
},
ratelimit=False,
event_dict = {
"type": type,
"room_id": room_id,
"sender": system_mxid,
"content": event_content,
}
if state_key is not None:
event_dict['state_key'] = state_key
res = yield self._event_creation_handler.create_and_send_nonmember_event(
requester, event_dict, ratelimit=False,
)
defer.returnValue(res)
@cachedInlineCallbacks()
def get_notice_room_for_user(self, user_id):
@ -141,6 +153,9 @@ class ServerNoticesManager(object):
creator_join_profile=join_profile,
)
room_id = info['room_id']
yield self._store.add_tag_to_room(
user_id, room_id, SERVER_NOTICE_ROOM_TAG, None
)
logger.info("Created server notices room %s for %s", room_id, user_id)
defer.returnValue(room_id)

View File

@ -12,7 +12,12 @@
# 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 twisted.internet import defer
from synapse.server_notices.consent_server_notices import ConsentServerNotices
from synapse.server_notices.resource_limits_server_notices import (
ResourceLimitsServerNotices,
)
class ServerNoticesSender(object):
@ -25,34 +30,34 @@ class ServerNoticesSender(object):
Args:
hs (synapse.server.HomeServer):
"""
# todo: it would be nice to make this more dynamic
self._consent_server_notices = ConsentServerNotices(hs)
self._server_notices = (
ConsentServerNotices(hs),
ResourceLimitsServerNotices(hs)
)
@defer.inlineCallbacks
def on_user_syncing(self, user_id):
"""Called when the user performs a sync operation.
Args:
user_id (str): mxid of user who synced
Returns:
Deferred
"""
return self._consent_server_notices.maybe_send_server_notice_to_user(
user_id,
)
for sn in self._server_notices:
yield sn.maybe_send_server_notice_to_user(
user_id,
)
@defer.inlineCallbacks
def on_user_ip(self, user_id):
"""Called on the master when a worker process saw a client request.
Args:
user_id (str): mxid
Returns:
Deferred
"""
# The synchrotrons use a stubbed version of ServerNoticesSender, so
# we check for notices to send to the user in on_user_ip as well as
# in on_user_syncing
return self._consent_server_notices.maybe_send_server_notice_to_user(
user_id,
)
for sn in self._server_notices:
yield sn.maybe_send_server_notice_to_user(
user_id,
)

View File

@ -147,6 +147,7 @@ class MonthlyActiveUsersStore(SQLBaseStore):
return count
return self.runInteraction("count_users", _count_users)
@defer.inlineCallbacks
def upsert_monthly_active_user(self, user_id):
"""
Updates or inserts monthly active user member
@ -155,7 +156,7 @@ class MonthlyActiveUsersStore(SQLBaseStore):
Deferred[bool]: True if a new entry was created, False if an
existing one was updated.
"""
is_insert = self._simple_upsert(
is_insert = yield self._simple_upsert(
desc="upsert_monthly_active_user",
table="monthly_active_users",
keyvalues={

View File

@ -458,7 +458,7 @@ class AuthTestCase(unittest.TestCase):
with self.assertRaises(ResourceLimitError) as e:
yield self.auth.check_auth_blocking()
self.assertEquals(e.exception.admin_uri, self.hs.config.admin_uri)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEED)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
self.assertEquals(e.exception.code, 403)
# Ensure does not throw an error
@ -474,5 +474,13 @@ class AuthTestCase(unittest.TestCase):
with self.assertRaises(ResourceLimitError) as e:
yield self.auth.check_auth_blocking()
self.assertEquals(e.exception.admin_uri, self.hs.config.admin_uri)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEED)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
self.assertEquals(e.exception.code, 403)
@defer.inlineCallbacks
def test_server_notices_mxid_special_cased(self):
self.hs.config.hs_disabled = True
user = "@user:server"
self.hs.config.server_notices_mxid = user
self.hs.config.hs_disabled_message = "Reason for being disabled"
yield self.auth.check_auth_blocking(user)

View File

@ -52,7 +52,7 @@ class SyncTestCase(tests.unittest.TestCase):
self.hs.config.hs_disabled = True
with self.assertRaises(ResourceLimitError) as e:
yield self.sync_handler.wait_for_sync_for_user(sync_config)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEED)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
self.hs.config.hs_disabled = False
@ -60,7 +60,7 @@ class SyncTestCase(tests.unittest.TestCase):
with self.assertRaises(ResourceLimitError) as e:
yield self.sync_handler.wait_for_sync_for_user(sync_config)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEED)
self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
def _generate_sync_config(self, user_id):
return SyncConfig(

View File

View File

@ -0,0 +1,212 @@
from mock import Mock
from twisted.internet import defer
from synapse.api.constants import EventTypes, ServerNoticeMsgType
from synapse.api.errors import ResourceLimitError
from synapse.handlers.auth import AuthHandler
from synapse.server_notices.resource_limits_server_notices import (
ResourceLimitsServerNotices,
)
from tests import unittest
from tests.utils import setup_test_homeserver
class AuthHandlers(object):
def __init__(self, hs):
self.auth_handler = AuthHandler(hs)
class TestResourceLimitsServerNotices(unittest.TestCase):
@defer.inlineCallbacks
def setUp(self):
self.hs = yield setup_test_homeserver(self.addCleanup, handlers=None)
self.hs.handlers = AuthHandlers(self.hs)
self.auth_handler = self.hs.handlers.auth_handler
self.server_notices_sender = self.hs.get_server_notices_sender()
# relying on [1] is far from ideal, but the only case where
# ResourceLimitsServerNotices class needs to be isolated is this test,
# general code should never have a reason to do so ...
self._rlsn = self.server_notices_sender._server_notices[1]
if not isinstance(self._rlsn, ResourceLimitsServerNotices):
raise Exception("Failed to find reference to ResourceLimitsServerNotices")
self._rlsn._store.user_last_seen_monthly_active = Mock(
return_value=defer.succeed(1000)
)
self._send_notice = self._rlsn._server_notices_manager.send_notice
self._rlsn._server_notices_manager.send_notice = Mock()
self._rlsn._state.get_current_state = Mock(return_value=defer.succeed(None))
self._rlsn._store.get_events = Mock(return_value=defer.succeed({}))
self._send_notice = self._rlsn._server_notices_manager.send_notice
self.hs.config.limit_usage_by_mau = True
self.user_id = "@user_id:test"
# self.server_notices_mxid = "@server:test"
# self.server_notices_mxid_display_name = None
# self.server_notices_mxid_avatar_url = None
# self.server_notices_room_name = "Server Notices"
self._rlsn._server_notices_manager.get_notice_room_for_user = Mock(
returnValue=""
)
self._rlsn._store.add_tag_to_room = Mock()
self.hs.config.admin_uri = "mailto:user@test.com"
@defer.inlineCallbacks
def test_maybe_send_server_notice_to_user_flag_off(self):
"""Tests cases where the flags indicate nothing to do"""
# test hs disabled case
self.hs.config.hs_disabled = True
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
self._send_notice.assert_not_called()
# Test when mau limiting disabled
self.hs.config.hs_disabled = False
self.hs.limit_usage_by_mau = False
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
self._send_notice.assert_not_called()
@defer.inlineCallbacks
def test_maybe_send_server_notice_to_user_remove_blocked_notice(self):
"""Test when user has blocked notice, but should have it removed"""
self._rlsn._auth.check_auth_blocking = Mock()
mock_event = Mock(
type=EventTypes.Message,
content={"msgtype": ServerNoticeMsgType},
)
self._rlsn._store.get_events = Mock(return_value=defer.succeed(
{"123": mock_event}
))
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
# Would be better to check the content, but once == remove blocking event
self._send_notice.assert_called_once()
@defer.inlineCallbacks
def test_maybe_send_server_notice_to_user_remove_blocked_notice_noop(self):
"""Test when user has blocked notice, but notice ought to be there (NOOP)"""
self._rlsn._auth.check_auth_blocking = Mock(
side_effect=ResourceLimitError(403, 'foo')
)
mock_event = Mock(
type=EventTypes.Message,
content={"msgtype": ServerNoticeMsgType},
)
self._rlsn._store.get_events = Mock(return_value=defer.succeed(
{"123": mock_event}
))
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
self._send_notice.assert_not_called()
@defer.inlineCallbacks
def test_maybe_send_server_notice_to_user_add_blocked_notice(self):
"""Test when user does not have blocked notice, but should have one"""
self._rlsn._auth.check_auth_blocking = Mock(
side_effect=ResourceLimitError(403, 'foo')
)
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
# Would be better to check contents, but 2 calls == set blocking event
self.assertTrue(self._send_notice.call_count == 2)
@defer.inlineCallbacks
def test_maybe_send_server_notice_to_user_add_blocked_notice_noop(self):
"""Test when user does not have blocked notice, nor should they (NOOP)"""
self._rlsn._auth.check_auth_blocking = Mock()
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
self._send_notice.assert_not_called()
@defer.inlineCallbacks
def test_maybe_send_server_notice_to_user_not_in_mau_cohort(self):
"""Test when user is not part of the MAU cohort - this should not ever
happen - but ...
"""
self._rlsn._auth.check_auth_blocking = Mock()
self._rlsn._store.user_last_seen_monthly_active = Mock(
return_value=defer.succeed(None)
)
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
self._send_notice.assert_not_called()
class TestResourceLimitsServerNoticesWithRealRooms(unittest.TestCase):
@defer.inlineCallbacks
def setUp(self):
self.hs = yield setup_test_homeserver(self.addCleanup)
self.store = self.hs.get_datastore()
self.server_notices_sender = self.hs.get_server_notices_sender()
self.server_notices_manager = self.hs.get_server_notices_manager()
self.event_source = self.hs.get_event_sources()
# relying on [1] is far from ideal, but the only case where
# ResourceLimitsServerNotices class needs to be isolated is this test,
# general code should never have a reason to do so ...
self._rlsn = self.server_notices_sender._server_notices[1]
if not isinstance(self._rlsn, ResourceLimitsServerNotices):
raise Exception("Failed to find reference to ResourceLimitsServerNotices")
self.hs.config.limit_usage_by_mau = True
self.hs.config.hs_disabled = False
self.hs.config.max_mau_value = 5
self.hs.config.server_notices_mxid = "@server:test"
self.hs.config.server_notices_mxid_display_name = None
self.hs.config.server_notices_mxid_avatar_url = None
self.hs.config.server_notices_room_name = "Test Server Notice Room"
self.user_id = "@user_id:test"
self.hs.config.admin_uri = "mailto:user@test.com"
@defer.inlineCallbacks
def test_server_notice_only_sent_once(self):
self.store.get_monthly_active_count = Mock(
return_value=1000,
)
self.store.user_last_seen_monthly_active = Mock(
return_value=1000,
)
# Call the function multiple times to ensure we only send the notice once
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
yield self._rlsn.maybe_send_server_notice_to_user(self.user_id)
# Now lets get the last load of messages in the service notice room and
# check that there is only one server notice
room_id = yield self.server_notices_manager.get_notice_room_for_user(
self.user_id,
)
token = yield self.event_source.get_current_token()
events, _ = yield self.store.get_recent_events_for_room(
room_id, limit=100, end_token=token.room_key,
)
count = 0
for event in events:
if event.type != EventTypes.Message:
continue
if event.content.get("msgtype") != ServerNoticeMsgType:
continue
count += 1
self.assertEqual(count, 1)