test_event_auth.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2018 New Vector Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import unittest
  16. from synapse import event_auth
  17. from synapse.api.constants import RoomVersions
  18. from synapse.api.errors import AuthError
  19. from synapse.events import FrozenEvent
  20. class EventAuthTestCase(unittest.TestCase):
  21. def test_random_users_cannot_send_state_before_first_pl(self):
  22. """
  23. Check that, before the first PL lands, the creator is the only user
  24. that can send a state event.
  25. """
  26. creator = "@creator:example.com"
  27. joiner = "@joiner:example.com"
  28. auth_events = {
  29. ("m.room.create", ""): _create_event(creator),
  30. ("m.room.member", creator): _join_event(creator),
  31. ("m.room.member", joiner): _join_event(joiner),
  32. }
  33. # creator should be able to send state
  34. event_auth.check(
  35. RoomVersions.V1, _random_state_event(creator), auth_events,
  36. do_sig_check=False,
  37. )
  38. # joiner should not be able to send state
  39. self.assertRaises(
  40. AuthError,
  41. event_auth.check,
  42. RoomVersions.V1,
  43. _random_state_event(joiner),
  44. auth_events,
  45. do_sig_check=False,
  46. ),
  47. def test_state_default_level(self):
  48. """
  49. Check that users above the state_default level can send state and
  50. those below cannot
  51. """
  52. creator = "@creator:example.com"
  53. pleb = "@joiner:example.com"
  54. king = "@joiner2:example.com"
  55. auth_events = {
  56. ("m.room.create", ""): _create_event(creator),
  57. ("m.room.member", creator): _join_event(creator),
  58. ("m.room.power_levels", ""): _power_levels_event(
  59. creator, {"state_default": "30", "users": {pleb: "29", king: "30"}}
  60. ),
  61. ("m.room.member", pleb): _join_event(pleb),
  62. ("m.room.member", king): _join_event(king),
  63. }
  64. # pleb should not be able to send state
  65. self.assertRaises(
  66. AuthError,
  67. event_auth.check,
  68. RoomVersions.V1,
  69. _random_state_event(pleb),
  70. auth_events,
  71. do_sig_check=False,
  72. ),
  73. # king should be able to send state
  74. event_auth.check(
  75. RoomVersions.V1, _random_state_event(king), auth_events,
  76. do_sig_check=False,
  77. )
  78. # helpers for making events
  79. TEST_ROOM_ID = "!test:room"
  80. def _create_event(user_id):
  81. return FrozenEvent(
  82. {
  83. "room_id": TEST_ROOM_ID,
  84. "event_id": _get_event_id(),
  85. "type": "m.room.create",
  86. "sender": user_id,
  87. "content": {"creator": user_id},
  88. }
  89. )
  90. def _join_event(user_id):
  91. return FrozenEvent(
  92. {
  93. "room_id": TEST_ROOM_ID,
  94. "event_id": _get_event_id(),
  95. "type": "m.room.member",
  96. "sender": user_id,
  97. "state_key": user_id,
  98. "content": {"membership": "join"},
  99. }
  100. )
  101. def _power_levels_event(sender, content):
  102. return FrozenEvent(
  103. {
  104. "room_id": TEST_ROOM_ID,
  105. "event_id": _get_event_id(),
  106. "type": "m.room.power_levels",
  107. "sender": sender,
  108. "state_key": "",
  109. "content": content,
  110. }
  111. )
  112. def _random_state_event(sender):
  113. return FrozenEvent(
  114. {
  115. "room_id": TEST_ROOM_ID,
  116. "event_id": _get_event_id(),
  117. "type": "test.state",
  118. "sender": sender,
  119. "state_key": "",
  120. "content": {"membership": "join"},
  121. }
  122. )
  123. event_count = 0
  124. def _get_event_id():
  125. global event_count
  126. c = event_count
  127. event_count += 1
  128. return "!%i:example.com" % (c,)