Add constants for history visibility.

synapse/api/auth.py

@@ -23,7 +23,7 @@
 import synapse.types
 from synapse import event_auth
 from synapse.api.auth_blocking import AuthBlocking
-from synapse.api.constants import EventTypes, Membership
+from synapse.api.constants import EventTypes, HistoryVisibility, Membership
 from synapse.api.errors import (
     AuthError,
     Codes,
@@ -648,7 +648,8 @@ async def check_user_in_room_or_world_readable(
             )
             if (
                 visibility
-                and visibility.content.get("history_visibility") == "world_readable"
+                and visibility.content.get("history_visibility")
+                == HistoryVisibility.WORLD_READABLE
             ):
                 return Membership.JOIN, None
             raise AuthError(

synapse/api/constants.py

@@ -160,3 +160,10 @@ class RoomEncryptionAlgorithms:
 class AccountDataTypes:
     DIRECT = "m.direct"
     IGNORED_USER_LIST = "m.ignored_user_list"
+
+
+class HistoryVisibility:
+    INVITED = "invited"
+    JOINED = "joined"
+    SHARED = "shared"
+    WORLD_READABLE = "world_readable"

synapse/handlers/room.py

@@ -27,6 +27,7 @@
 
 from synapse.api.constants import (
     EventTypes,
+    HistoryVisibility,
     JoinRules,
     Membership,
     RoomCreationPreset,
@@ -81,21 +82,21 @@ def __init__(self, hs: "HomeServer"):
         self._presets_dict = {
             RoomCreationPreset.PRIVATE_CHAT: {
                 "join_rules": JoinRules.INVITE,
-                "history_visibility": "shared",
+                "history_visibility": HistoryVisibility.SHARED,
                 "original_invitees_have_ops": False,
                 "guest_can_join": True,
                 "power_level_content_override": {"invite": 0},
             },
             RoomCreationPreset.TRUSTED_PRIVATE_CHAT: {
                 "join_rules": JoinRules.INVITE,
-                "history_visibility": "shared",
+                "history_visibility": HistoryVisibility.SHARED,
                 "original_invitees_have_ops": True,
                 "guest_can_join": True,
                 "power_level_content_override": {"invite": 0},
             },
             RoomCreationPreset.PUBLIC_CHAT: {
                 "join_rules": JoinRules.PUBLIC,
-                "history_visibility": "shared",
+                "history_visibility": HistoryVisibility.SHARED,
                 "original_invitees_have_ops": False,
                 "guest_can_join": False,
                 "power_level_content_override": {},

synapse/handlers/room_list.py

@@ -20,7 +20,7 @@
 import msgpack
 from unpaddedbase64 import decode_base64, encode_base64
 
-from synapse.api.constants import EventTypes, JoinRules
+from synapse.api.constants import EventTypes, HistoryVisibility, JoinRules
 from synapse.api.errors import Codes, HttpResponseException
 from synapse.types import ThirdPartyInstanceID
 from synapse.util.caches.descriptors import cached
@@ -159,7 +159,8 @@ def build_room_entry(room):
                 "canonical_alias": room["canonical_alias"],
                 "num_joined_members": room["joined_members"],
                 "avatar_url": room["avatar"],
-                "world_readable": room["history_visibility"] == "world_readable",
+                "world_readable": room["history_visibility"]
+                == HistoryVisibility.WORLD_READABLE,
                 "guest_can_join": room["guest_access"] == "can_join",
             }
 
@@ -317,7 +318,7 @@ async def generate_room_entry(
         visibility = None
         if visibility_event:
             visibility = visibility_event.content.get("history_visibility", None)
-        result["world_readable"] = visibility == "world_readable"
+        result["world_readable"] = visibility == HistoryVisibility.WORLD_READABLE
 
         guest_event = current_state.get((EventTypes.GuestAccess, ""))
         guest = None

synapse/handlers/user_directory.py

@@ -16,7 +16,7 @@
 import logging
 
 import synapse.metrics
-from synapse.api.constants import EventTypes, JoinRules, Membership
+from synapse.api.constants import EventTypes, HistoryVisibility, JoinRules, Membership
 from synapse.handlers.state_deltas import StateDeltasHandler
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.roommember import ProfileInfo
@@ -250,7 +250,7 @@ async def _handle_room_publicity_change(
                 prev_event_id,
                 event_id,
                 key_name="history_visibility",
-                public_value="world_readable",
+                public_value=HistoryVisibility.WORLD_READABLE,
             )
         elif typ == EventTypes.JoinRules:
             change = await self._get_key_change(

synapse/notifier.py

@@ -34,7 +34,7 @@
 from twisted.internet import defer
 
 import synapse.server
-from synapse.api.constants import EventTypes, Membership
+from synapse.api.constants import EventTypes, HistoryVisibility, Membership
 from synapse.api.errors import AuthError
 from synapse.events import EventBase
 from synapse.handlers.presence import format_user_presence_state
@@ -611,7 +611,9 @@ async def _is_world_readable(self, room_id: str) -> bool:
             room_id, EventTypes.RoomHistoryVisibility, ""
         )
         if state and "history_visibility" in state.content:
-            return state.content["history_visibility"] == "world_readable"
+            return (
+                state.content["history_visibility"] == HistoryVisibility.WORLD_READABLE
+            )
         else:
             return False
 

synapse/storage/databases/main/user_directory.py

@@ -17,7 +17,7 @@
 import re
 from typing import Any, Dict, Iterable, Optional, Set, Tuple
 
-from synapse.api.constants import EventTypes, JoinRules
+from synapse.api.constants import EventTypes, HistoryVisibility, JoinRules
 from synapse.storage.database import DatabasePool
 from synapse.storage.databases.main.state import StateFilter
 from synapse.storage.databases.main.state_deltas import StateDeltasStore
@@ -360,7 +360,10 @@ async def is_room_world_readable_or_publicly_joinable(self, room_id):
         if hist_vis_id:
             hist_vis_ev = await self.get_event(hist_vis_id, allow_none=True)
             if hist_vis_ev:
-                if hist_vis_ev.content.get("history_visibility") == "world_readable":
+                if (
+                    hist_vis_ev.content.get("history_visibility")
+                    == HistoryVisibility.WORLD_READABLE
+                ):
                     return True
 
         return False

synapse/visibility.py

@@ -12,11 +12,15 @@
 # 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
 import operator
 
-from synapse.api.constants import AccountDataTypes, EventTypes, Membership
+from synapse.api.constants import (
+    AccountDataTypes,
+    EventTypes,
+    HistoryVisibility,
+    Membership,
+)
 from synapse.events.utils import prune_event
 from synapse.storage import Storage
 from synapse.storage.state import StateFilter
@@ -25,7 +29,12 @@
 logger = logging.getLogger(__name__)
 
 
-VISIBILITY_PRIORITY = ("world_readable", "shared", "invited", "joined")
+VISIBILITY_PRIORITY = (
+    HistoryVisibility.WORLD_READABLE,
+    HistoryVisibility.SHARED,
+    HistoryVisibility.INVITED,
+    HistoryVisibility.JOINED,
+)
 
 
 MEMBERSHIP_PRIORITY = (
@@ -150,12 +159,14 @@ def allowed(event):
         # get the room_visibility at the time of the event.
         visibility_event = state.get((EventTypes.RoomHistoryVisibility, ""), None)
         if visibility_event:
-            visibility = visibility_event.content.get("history_visibility", "shared")
+            visibility = visibility_event.content.get(
+                "history_visibility", HistoryVisibility.SHARED
+            )
         else:
-            visibility = "shared"
+            visibility = HistoryVisibility.SHARED
 
         if visibility not in VISIBILITY_PRIORITY:
-            visibility = "shared"
+            visibility = HistoryVisibility.SHARED
 
         # Always allow history visibility events on boundaries. This is done
         # by setting the effective visibility to the least restrictive
@@ -165,7 +176,7 @@ def allowed(event):
             prev_visibility = prev_content.get("history_visibility", None)
 
             if prev_visibility not in VISIBILITY_PRIORITY:
-                prev_visibility = "shared"
+                prev_visibility = HistoryVisibility.SHARED
 
             new_priority = VISIBILITY_PRIORITY.index(visibility)
             old_priority = VISIBILITY_PRIORITY.index(prev_visibility)
@@ -210,17 +221,17 @@ def allowed(event):
 
         # otherwise, it depends on the room visibility.
 
-        if visibility == "joined":
+        if visibility == HistoryVisibility.JOINED:
             # we weren't a member at the time of the event, so we can't
             # see this event.
             return None
 
-        elif visibility == "invited":
+        elif visibility == HistoryVisibility.INVITED:
             # user can also see the event if they were *invited* at the time
             # of the event.
             return event if membership == Membership.INVITE else None
 
-        elif visibility == "shared" and is_peeking:
+        elif visibility == HistoryVisibility.SHARED and is_peeking:
             # if the visibility is shared, users cannot see the event unless
             # they have *subequently* joined the room (or were members at the
             # time, of course)
@@ -284,8 +295,10 @@ def is_sender_erased(event, erased_senders):
     def check_event_is_visible(event, state):
         history = state.get((EventTypes.RoomHistoryVisibility, ""), None)
         if history:
-            visibility = history.content.get("history_visibility", "shared")
-            if visibility in ["invited", "joined"]:
+            visibility = history.content.get(
+                "history_visibility", HistoryVisibility.SHARED
+            )
+            if visibility in [HistoryVisibility.INVITED, HistoryVisibility.JOINED]:
                 # We now loop through all state events looking for
                 # membership states for the requesting server to determine
                 # if the server is either in the room or has been invited
@@ -305,7 +318,7 @@ def check_event_is_visible(event, state):
                     if memtype == Membership.JOIN:
                         return True
                     elif memtype == Membership.INVITE:
-                        if visibility == "invited":
+                        if visibility == HistoryVisibility.INVITED:
                             return True
                 else:
                     # server has no users in the room: redact
@@ -336,7 +349,8 @@ def check_event_is_visible(event, state):
     else:
         event_map = await storage.main.get_events(visibility_ids)
         all_open = all(
-            e.content.get("history_visibility") in (None, "shared", "world_readable")
+            e.content.get("history_visibility")
+            in (None, HistoryVisibility.SHARED, HistoryVisibility.WORLD_READABLE)
             for e in event_map.values()
         )