test_event_auth.py

# -*- 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 unittest

from synapse import event_auth
from synapse.api.errors import AuthError
from synapse.api.room_versions import RoomVersions
from synapse.events import make_event_from_dict
from synapse.types import get_domain_from_id


class EventAuthTestCase(unittest.TestCase):
    def test_random_users_cannot_send_state_before_first_pl(self):
        """
        Check that, before the first PL lands, the creator is the only user
        that can send a state event.
        """
        creator = "@creator:example.com"
        joiner = "@joiner:example.com"
        auth_events = {
            ("m.room.create", ""): _create_event(creator),
            ("m.room.member", creator): _join_event(creator),
            ("m.room.member", joiner): _join_event(joiner),
        }

        # creator should be able to send state
        event_auth.check(
            RoomVersions.V1,
            _random_state_event(creator),
            auth_events,
            do_sig_check=False,
        )

        # joiner should not be able to send state
        self.assertRaises(
            AuthError,
            event_auth.check,
            RoomVersions.V1,
            _random_state_event(joiner),
            auth_events,
            do_sig_check=False,
        )

    def test_state_default_level(self):
        """
        Check that users above the state_default level can send state and
        those below cannot
        """
        creator = "@creator:example.com"
        pleb = "@joiner:example.com"
        king = "@joiner2:example.com"

        auth_events = {
            ("m.room.create", ""): _create_event(creator),
            ("m.room.member", creator): _join_event(creator),
            ("m.room.power_levels", ""): _power_levels_event(
                creator, {"state_default": "30", "users": {pleb: "29", king: "30"}}
            ),
            ("m.room.member", pleb): _join_event(pleb),
            ("m.room.member", king): _join_event(king),
        }

        # pleb should not be able to send state
        self.assertRaises(
            AuthError,
            event_auth.check,
            RoomVersions.V1,
            _random_state_event(pleb),
            auth_events,
            do_sig_check=False,
        ),

        # king should be able to send state
        event_auth.check(
            RoomVersions.V1, _random_state_event(king), auth_events, do_sig_check=False,
        )

    def test_alias_event(self):
        """Alias events have special behavior up through room version 6."""
        creator = "@creator:example.com"
        other = "@other:example.com"
        auth_events = {
            ("m.room.create", ""): _create_event(creator),
            ("m.room.member", creator): _join_event(creator),
        }

        # creator should be able to send aliases
        event_auth.check(
            RoomVersions.V1, _alias_event(creator), auth_events, do_sig_check=False,
        )

        # Reject an event with no state key.
        with self.assertRaises(AuthError):
            event_auth.check(
                RoomVersions.V1,
                _alias_event(creator, state_key=""),
                auth_events,
                do_sig_check=False,
            )

        # If the domain of the sender does not match the state key, reject.
        with self.assertRaises(AuthError):
            event_auth.check(
                RoomVersions.V1,
                _alias_event(creator, state_key="test.com"),
                auth_events,
                do_sig_check=False,
            )

        # Note that the member does *not* need to be in the room.
        event_auth.check(
            RoomVersions.V1, _alias_event(other), auth_events, do_sig_check=False,
        )

    def test_msc2432_alias_event(self):
        """After MSC2432, alias events have no special behavior."""
        creator = "@creator:example.com"
        other = "@other:example.com"
        auth_events = {
            ("m.room.create", ""): _create_event(creator),
            ("m.room.member", creator): _join_event(creator),
        }

        # creator should be able to send aliases
        event_auth.check(
            RoomVersions.MSC2432_DEV,
            _alias_event(creator),
            auth_events,
            do_sig_check=False,
        )

        # No particular checks are done on the state key.
        event_auth.check(
            RoomVersions.MSC2432_DEV,
            _alias_event(creator, state_key=""),
            auth_events,
            do_sig_check=False,
        )
        event_auth.check(
            RoomVersions.MSC2432_DEV,
            _alias_event(creator, state_key="test.com"),
            auth_events,
            do_sig_check=False,
        )

        # Per standard auth rules, the member must be in the room.
        with self.assertRaises(AuthError):
            event_auth.check(
                RoomVersions.MSC2432_DEV,
                _alias_event(other),
                auth_events,
                do_sig_check=False,
            )


# helpers for making events

TEST_ROOM_ID = "!test:room"


def _create_event(user_id):
    return make_event_from_dict(
        {
            "room_id": TEST_ROOM_ID,
            "event_id": _get_event_id(),
            "type": "m.room.create",
            "sender": user_id,
            "content": {"creator": user_id},
        }
    )


def _join_event(user_id):
    return make_event_from_dict(
        {
            "room_id": TEST_ROOM_ID,
            "event_id": _get_event_id(),
            "type": "m.room.member",
            "sender": user_id,
            "state_key": user_id,
            "content": {"membership": "join"},
        }
    )


def _power_levels_event(sender, content):
    return make_event_from_dict(
        {
            "room_id": TEST_ROOM_ID,
            "event_id": _get_event_id(),
            "type": "m.room.power_levels",
            "sender": sender,
            "state_key": "",
            "content": content,
        }
    )


def _alias_event(sender, **kwargs):
    data = {
        "room_id": TEST_ROOM_ID,
        "event_id": _get_event_id(),
        "type": "m.room.aliases",
        "sender": sender,
        "state_key": get_domain_from_id(sender),
        "content": {"aliases": []},
    }
    data.update(**kwargs)
    return make_event_from_dict(data)


def _random_state_event(sender):
    return make_event_from_dict(
        {
            "room_id": TEST_ROOM_ID,
            "event_id": _get_event_id(),
            "type": "test.state",
            "sender": sender,
            "state_key": "",
            "content": {"membership": "join"},
        }
    )


event_count = 0


def _get_event_id():
    global event_count
    c = event_count
    event_count += 1
    return "!%i:example.com" % (c,)