1
0

test_event_auth.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157
  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.errors import AuthError
  18. from synapse.api.room_versions import RoomVersions
  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.identifier,
  36. _random_state_event(creator),
  37. auth_events,
  38. do_sig_check=False,
  39. )
  40. # joiner should not be able to send state
  41. self.assertRaises(
  42. AuthError,
  43. event_auth.check,
  44. RoomVersions.V1.identifier,
  45. _random_state_event(joiner),
  46. auth_events,
  47. do_sig_check=False,
  48. ),
  49. def test_state_default_level(self):
  50. """
  51. Check that users above the state_default level can send state and
  52. those below cannot
  53. """
  54. creator = "@creator:example.com"
  55. pleb = "@joiner:example.com"
  56. king = "@joiner2:example.com"
  57. auth_events = {
  58. ("m.room.create", ""): _create_event(creator),
  59. ("m.room.member", creator): _join_event(creator),
  60. ("m.room.power_levels", ""): _power_levels_event(
  61. creator, {"state_default": "30", "users": {pleb: "29", king: "30"}}
  62. ),
  63. ("m.room.member", pleb): _join_event(pleb),
  64. ("m.room.member", king): _join_event(king),
  65. }
  66. # pleb should not be able to send state
  67. self.assertRaises(
  68. AuthError,
  69. event_auth.check,
  70. RoomVersions.V1.identifier,
  71. _random_state_event(pleb),
  72. auth_events,
  73. do_sig_check=False,
  74. ),
  75. # king should be able to send state
  76. event_auth.check(
  77. RoomVersions.V1.identifier,
  78. _random_state_event(king),
  79. auth_events,
  80. do_sig_check=False,
  81. )
  82. # helpers for making events
  83. TEST_ROOM_ID = "!test:room"
  84. def _create_event(user_id):
  85. return FrozenEvent(
  86. {
  87. "room_id": TEST_ROOM_ID,
  88. "event_id": _get_event_id(),
  89. "type": "m.room.create",
  90. "sender": user_id,
  91. "content": {"creator": user_id},
  92. }
  93. )
  94. def _join_event(user_id):
  95. return FrozenEvent(
  96. {
  97. "room_id": TEST_ROOM_ID,
  98. "event_id": _get_event_id(),
  99. "type": "m.room.member",
  100. "sender": user_id,
  101. "state_key": user_id,
  102. "content": {"membership": "join"},
  103. }
  104. )
  105. def _power_levels_event(sender, content):
  106. return FrozenEvent(
  107. {
  108. "room_id": TEST_ROOM_ID,
  109. "event_id": _get_event_id(),
  110. "type": "m.room.power_levels",
  111. "sender": sender,
  112. "state_key": "",
  113. "content": content,
  114. }
  115. )
  116. def _random_state_event(sender):
  117. return FrozenEvent(
  118. {
  119. "room_id": TEST_ROOM_ID,
  120. "event_id": _get_event_id(),
  121. "type": "test.state",
  122. "sender": sender,
  123. "state_key": "",
  124. "content": {"membership": "join"},
  125. }
  126. )
  127. event_count = 0
  128. def _get_event_id():
  129. global event_count
  130. c = event_count
  131. event_count += 1
  132. return "!%i:example.com" % (c,)