test_event_auth.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  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.events import FrozenEvent
  19. class EventAuthTestCase(unittest.TestCase):
  20. def test_random_users_cannot_send_state_before_first_pl(self):
  21. """
  22. Check that, before the first PL lands, the creator is the only user
  23. that can send a state event.
  24. """
  25. creator = "@creator:example.com"
  26. joiner = "@joiner:example.com"
  27. auth_events = {
  28. ("m.room.create", ""): _create_event(creator),
  29. ("m.room.member", creator): _join_event(creator),
  30. ("m.room.member", joiner): _join_event(joiner),
  31. }
  32. # creator should be able to send state
  33. event_auth.check(
  34. _random_state_event(creator), auth_events,
  35. do_sig_check=False,
  36. )
  37. # joiner should not be able to send state
  38. self.assertRaises(
  39. AuthError,
  40. event_auth.check,
  41. _random_state_event(joiner),
  42. auth_events,
  43. do_sig_check=False,
  44. ),
  45. def test_state_default_level(self):
  46. """
  47. Check that users above the state_default level can send state and
  48. those below cannot
  49. """
  50. creator = "@creator:example.com"
  51. pleb = "@joiner:example.com"
  52. king = "@joiner2:example.com"
  53. auth_events = {
  54. ("m.room.create", ""): _create_event(creator),
  55. ("m.room.member", creator): _join_event(creator),
  56. ("m.room.power_levels", ""): _power_levels_event(creator, {
  57. "state_default": "30",
  58. "users": {
  59. pleb: "29",
  60. king: "30",
  61. },
  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. _random_state_event(pleb),
  71. auth_events,
  72. do_sig_check=False,
  73. ),
  74. # king should be able to send state
  75. event_auth.check(
  76. _random_state_event(king), auth_events,
  77. do_sig_check=False,
  78. )
  79. # helpers for making events
  80. TEST_ROOM_ID = "!test:room"
  81. def _create_event(user_id):
  82. return FrozenEvent({
  83. "room_id": TEST_ROOM_ID,
  84. "event_id": _get_event_id(),
  85. "type": "m.room.create",
  86. "sender": user_id,
  87. "content": {
  88. "creator": user_id,
  89. },
  90. })
  91. def _join_event(user_id):
  92. return FrozenEvent({
  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": {
  99. "membership": "join",
  100. },
  101. })
  102. def _power_levels_event(sender, content):
  103. return FrozenEvent({
  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. def _random_state_event(sender):
  112. return FrozenEvent({
  113. "room_id": TEST_ROOM_ID,
  114. "event_id": _get_event_id(),
  115. "type": "test.state",
  116. "sender": sender,
  117. "state_key": "",
  118. "content": {
  119. "membership": "join",
  120. },
  121. })
  122. event_count = 0
  123. def _get_event_id():
  124. global event_count
  125. c = event_count
  126. event_count += 1
  127. return "!%i:example.com" % (c, )