message_fetch: Allow access to web-public msgs for unauth users.

Via API, users can now access messages which are in web-public
streams without any authentication.

If the user is not authenticated, we assume it is a web-public
query and add `streams:web-public` narrow if not already present
to the narrow. web-public streams are also directly accessible.

Any malformed narrow which is not allowed in a web-public query
results in a 400 or 401. See test_message_fetch for the allowed
queries.

zerver/lib/narrow.py

@@ -23,7 +23,7 @@ def check_supported_events_narrow_filter(narrow: Iterable[Sequence[str]]) -> Non
         if operator not in ["stream", "topic", "sender", "is"]:
             raise JsonableError(_("Operator {} not supported.").format(operator))
 
-def is_web_public_compatible(narrow: Iterable[Dict[str, str]]) -> bool:
+def is_web_public_compatible(narrow: Iterable[Dict[str, Any]]) -> bool:
     for element in narrow:
         operator = element['operator']
         if 'operand' not in element:
@@ -32,6 +32,18 @@ def is_web_public_compatible(narrow: Iterable[Dict[str, str]]) -> bool:
             return False
     return True
 
+def is_web_public_narrow(narrow: Optional[Iterable[Dict[str, Any]]]) -> bool:
+    if narrow is None:
+        return False
+
+    for term in narrow:
+        # Web public queries are only allowed for limited types of narrows.
+        # term == {'operator': 'streams', 'operand': 'web-public', 'negated': False}
+        if term['operator'] == 'streams' and term['operand'] == 'web-public' and term['negated'] is False:
+            return True
+
+    return False
+
 def build_narrow_filter(narrow: Iterable[Sequence[str]]) -> Callable[[Mapping[str, Any]], bool]:
     """Changes to this function should come with corresponding changes to
     BuildNarrowFilterTest."""

zerver/lib/streams.py

@@ -285,6 +285,13 @@ def get_public_streams_queryset(realm: Realm) -> 'QuerySet[Stream]':
     return Stream.objects.filter(realm=realm, invite_only=False,
                                  history_public_to_subscribers=True)
 
+def get_web_public_streams_queryset(realm: Realm) -> 'QuerySet[Stream]':
+    # In theory, is_web_public=True implies invite_only=False and
+    # history_public_to_subscribers=True, but it's safer to include
+    # this in the query.
+    return Stream.objects.filter(realm=realm, deactivated=False, invite_only=False,
+                                 history_public_to_subscribers=True, is_web_public=True)
+
 def get_stream_by_id(stream_id: int) -> Stream:
     error = _("Invalid stream id")
     try:

zerver/tests/test_message_fetch.py

@@ -5,6 +5,7 @@ from unittest import mock
 
 import orjson
 from django.db import connection
+from django.http import HttpResponse
 from django.test import override_settings
 from django.utils.timezone import now as timezone_now
 from sqlalchemy.sql import and_, column, select, table
@@ -18,6 +19,7 @@ from zerver.lib.actions import (
     do_set_realm_property,
     do_update_message,
 )
+from zerver.lib.avatar import avatar_url
 from zerver.lib.markdown import MentionData
 from zerver.lib.message import (
     MessageDict,
@@ -93,7 +95,7 @@ class NarrowBuilderTest(ZulipTestCase):
         super().setUp()
         self.realm = get_realm('zulip')
         self.user_profile = self.example_user('hamlet')
-        self.builder = NarrowBuilder(self.user_profile, column('id'))
+        self.builder = NarrowBuilder(self.user_profile, column('id'), self.realm)
         self.raw_query = select([column("id")], None, table("zerver_message"))
         self.hamlet_email = self.example_user('hamlet').email
         self.othello_email = self.example_user('othello').email
@@ -447,6 +449,16 @@ class NarrowBuilderTest(ZulipTestCase):
         query = self._build_query(term)
         self.assertEqual(get_sqlalchemy_sql(query), 'SELECT id \nFROM zerver_message')
 
+    def test_add_term_non_web_public_stream_in_web_public_query(self) -> None:
+        self.make_stream('non-web-public-stream', realm=self.realm)
+        term = dict(operator='stream', operand='non-web-public-stream')
+        builder = NarrowBuilder(self.user_profile, column('id'), self.realm, True)
+
+        def _build_query(term: Dict[str, Any]) -> Query:
+            return builder.add_term(self.raw_query, term)
+
+        self.assertRaises(BadNarrowOperator, _build_query, term)
+
     def _do_add_term_test(self, term: Dict[str, Any], where_clause: str,
                           params: Optional[Dict[str, Any]]=None) -> None:
         query = self._build_query(term)
@@ -523,19 +535,19 @@ class IncludeHistoryTest(ZulipTestCase):
         narrow = [
             dict(operator='stream', operand='public_stream', negated=True),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # streams:public searches should include history for non-guest members.
         narrow = [
             dict(operator='streams', operand='public'),
         ]
-        self.assertTrue(ok_to_include_history(narrow, user_profile))
+        self.assertTrue(ok_to_include_history(narrow, user_profile, False))
 
         # Negated -streams:public searches should not include history.
         narrow = [
             dict(operator='streams', operand='public', negated=True),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # Definitely forbid seeing history on private streams.
         self.make_stream('private_stream', realm=user_profile.realm, invite_only=True)
@@ -544,7 +556,7 @@ class IncludeHistoryTest(ZulipTestCase):
         narrow = [
             dict(operator='stream', operand='private_stream'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # Verify that with stream.history_public_to_subscribers, subscribed
         # users can access history.
@@ -555,20 +567,20 @@ class IncludeHistoryTest(ZulipTestCase):
         narrow = [
             dict(operator='stream', operand='private_stream_2'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
-        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile))
+        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
 
         # History doesn't apply to PMs.
         narrow = [
             dict(operator='is', operand='private'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # History doesn't apply to unread messages.
         narrow = [
             dict(operator='is', operand='unread'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # If we are looking for something like starred messages, there is
         # no point in searching historical messages.
@@ -576,7 +588,7 @@ class IncludeHistoryTest(ZulipTestCase):
             dict(operator='stream', operand='public_stream'),
             dict(operator='is', operand='starred'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # No point in searching history for is operator even if included with
         # streams:public
@@ -584,30 +596,30 @@ class IncludeHistoryTest(ZulipTestCase):
             dict(operator='streams', operand='public'),
             dict(operator='is', operand='mentioned'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
         narrow = [
             dict(operator='streams', operand='public'),
             dict(operator='is', operand='unread'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
         narrow = [
             dict(operator='streams', operand='public'),
             dict(operator='is', operand='alerted'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, user_profile))
+        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
 
         # simple True case
         narrow = [
             dict(operator='stream', operand='public_stream'),
         ]
-        self.assertTrue(ok_to_include_history(narrow, user_profile))
+        self.assertTrue(ok_to_include_history(narrow, user_profile, False))
 
         narrow = [
             dict(operator='stream', operand='public_stream'),
             dict(operator='topic', operand='whatever'),
             dict(operator='search', operand='needle in haystack'),
         ]
-        self.assertTrue(ok_to_include_history(narrow, user_profile))
+        self.assertTrue(ok_to_include_history(narrow, user_profile, False))
 
         # Tests for guest user
         guest_user_profile = self.example_user("polonius")
@@ -618,23 +630,23 @@ class IncludeHistoryTest(ZulipTestCase):
         narrow = [
             dict(operator='streams', operand='public'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, guest_user_profile))
+        self.assertFalse(ok_to_include_history(narrow, guest_user_profile, False))
 
         # Guest user can't access public stream
         self.subscribe(subscribed_user_profile, 'public_stream_2')
         narrow = [
             dict(operator='stream', operand='public_stream_2'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, guest_user_profile))
+        self.assertFalse(ok_to_include_history(narrow, guest_user_profile, False))
-        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile))
+        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
 
         # Definitely, a guest user can't access the unsubscribed private stream
         self.subscribe(subscribed_user_profile, 'private_stream_3')
         narrow = [
             dict(operator='stream', operand='private_stream_3'),
         ]
-        self.assertFalse(ok_to_include_history(narrow, guest_user_profile))
+        self.assertFalse(ok_to_include_history(narrow, guest_user_profile, False))
-        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile))
+        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
 
         # Guest user can access (history of) subscribed private streams
         self.subscribe(guest_user_profile, 'private_stream_4')
@@ -642,8 +654,8 @@ class IncludeHistoryTest(ZulipTestCase):
         narrow = [
             dict(operator='stream', operand='private_stream_4'),
         ]
-        self.assertTrue(ok_to_include_history(narrow, guest_user_profile))
+        self.assertTrue(ok_to_include_history(narrow, guest_user_profile, False))
-        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile))
+        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
 
 class PostProcessTest(ZulipTestCase):
     def test_basics(self) -> None:
@@ -1192,6 +1204,157 @@ class GetOldMessagesTest(ZulipTestCase):
             ),
         )
 
+    def test_unauthenticated_get_messages_non_existant_realm(self) -> None:
+        post_params = {
+            "anchor": 10000000000000000,
+            "num_before": 5,
+            "num_after": 1,
+            "narrow": orjson.dumps([dict(operator='streams', operand="web-public")]).decode(),
+        }
+
+        with mock.patch('zerver.views.message_fetch.get_realm_from_request', return_value=None):
+            result = self.client_get("/json/messages", dict(post_params))
+            self.assert_json_error(result, "Invalid subdomain.",
+                                   status_code=400)
+
+    def test_unauthenticated_get_messages_without_web_public(self) -> None:
+        """
+        An unauthenticated call to GET /json/messages with valid parameters
+        returns a 401.
+        """
+        post_params = {
+            "anchor": 1,
+            "num_before": 1,
+            "num_after": 1,
+            "narrow": orjson.dumps([dict(operator='is', operand="private")]).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(result, "Not logged in: API authentication or user session required",
+                               status_code=401)
+
+        post_params = {
+            "anchor": 10000000000000000,
+            "num_before": 5,
+            "num_after": 1,
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(result, "Not logged in: API authentication or user session required",
+                               status_code=401)
+
+    def test_unauthenticated_get_messages_with_web_public(self) -> None:
+        """
+        An unauthenticated call to GET /json/messages without valid
+        parameters in the `streams:web-public` narrow returns a 401.
+        """
+        post_params: Dict[str, Union[int, str, bool]] = {
+            "anchor": 1,
+            "num_before": 1,
+            "num_after": 1,
+            # "is:private" is not a is_web_public_compatible narrow.
+            "narrow": orjson.dumps([dict(operator="streams", operand="web-public"),
+                                    dict(operator="is", operand="private")]).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(result, 'Not logged in: API authentication or user session required',
+                               status_code=401)
+
+    def test_unauthenticated_narrow_to_non_web_public_streams_without_web_public(self) -> None:
+        """
+        An unauthenticated call to GET /json/messages without `streams:web-public` narrow returns a 401.
+        """
+        post_params: Dict[str, Union[int, str, bool]] = {
+            "anchor": 1,
+            "num_before": 1,
+            "num_after": 1,
+            "narrow": orjson.dumps([dict(operator='stream', operand='Scotland')]).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(result, "Not logged in: API authentication or user session required",
+                               status_code=401)
+
+    def test_unauthenticated_narrow_to_non_web_public_streams_with_web_public(self) -> None:
+        """
+        An unauthenticated call to GET /json/messages with valid
+        parameters in the `streams:web-public` narrow + narrow to stream returns
+        a 400 if the target stream is not web-public.
+        """
+        post_params: Dict[str, Union[int, str, bool]] = {
+            "anchor": 1,
+            "num_before": 1,
+            "num_after": 1,
+            "narrow": orjson.dumps([dict(operator="streams", operand="web-public"),
+                                    dict(operator='stream', operand='Scotland')]).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.assert_json_error(result, 'Invalid narrow operator: unknown web-public stream Scotland',
+                               status_code=400)
+
+    def setup_web_public_test(self, num_web_public_message: int=1) -> None:
+        """
+        Send N+2 messages, N in a web-public stream, then one in a non web-public stream
+        and then a private message.
+        """
+        user_profile = self.example_user('iago')
+        self.login('iago')
+        web_public_stream = self.make_stream('web-public-stream', is_web_public=True)
+        non_web_public_stream = self.make_stream('non-web-public-stream')
+        self.subscribe(user_profile, web_public_stream.name)
+        self.subscribe(user_profile, non_web_public_stream.name)
+
+        for _ in range(num_web_public_message):
+            self.send_stream_message(user_profile, web_public_stream.name,
+                                     content="web-public message")
+        self.send_stream_message(user_profile, non_web_public_stream.name,
+                                 content="non web-public message")
+        self.send_personal_message(user_profile, self.example_user('hamlet'),
+                                   content="private message")
+        self.logout()
+
+    def verify_web_public_query_result_success(self, result: HttpResponse, expected_num_messages: int) -> None:
+        self.assert_json_success(result)
+        messages = orjson.loads(result.content)['messages']
+        self.assert_length(messages, expected_num_messages)
+        sender = self.example_user('iago')
+        for msg in messages:
+            self.assertEqual(msg['content'], '<p>web-public message</p>')
+            self.assertEqual(msg['flags'], ['read'])
+            self.assertEqual(msg['sender_email'], sender.email)
+            self.assertEqual(msg['avatar_url'], avatar_url(sender))
+
+    def test_unauthenticated_narrow_to_web_public_streams(self) -> None:
+        self.setup_web_public_test()
+
+        post_params: Dict[str, Union[int, str, bool]] = {
+            "anchor": 1,
+            "num_before": 1,
+            "num_after": 1,
+            "narrow": orjson.dumps([dict(operator="streams", operand="web-public"),
+                                    dict(operator='stream', operand='web-public-stream')]).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        self.verify_web_public_query_result_success(result, 1)
+
+    def test_get_messages_with_web_public(self) -> None:
+        """
+        An unauthenticated call to GET /json/messages with valid parameters
+        including `streams:web-public` narrow returns list of messages in the
+        `web-public` streams.
+        """
+        self.setup_web_public_test(num_web_public_message=8)
+
+        post_params = {
+            "anchor": "first_unread",
+            "num_before": 5,
+            "num_after": 1,
+            "narrow": orjson.dumps([dict(operator='streams', operand="web-public")]).decode(),
+        }
+        result = self.client_get("/json/messages", dict(post_params))
+        # Of the last 7 (num_before + num_after + 1) messages, only 5
+        # messages are returned, which were all web-public messages.
+        # The other two messages should not be returned even though
+        # they are the most recent.
+        self.verify_web_public_query_result_success(result, 5)
+
     def test_client_avatar(self) -> None:
         """
         The client_gravatar flag determines whether we send avatar_url.

zerver/tests/test_urls.py

@@ -37,10 +37,11 @@ class PublicURLTest(ZulipTestCase):
                           "/en/accounts/login/", "/ru/accounts/login/",
                           "/help/"],
                     302: ["/", "/en/", "/ru/"],
+                    400: ["/json/messages",
+                          ],
                     401: [f"/json/streams/{denmark_stream_id}/members",
                           "/api/v1/users/me/subscriptions",
                           "/api/v1/messages",
-                          "/json/messages",
                           "/api/v1/streams",
                           ],
                     404: ["/help/nonexistent", "/help/include/admin",

zerver/views/message_fetch.py

@@ -3,6 +3,7 @@ from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 
 import orjson
 from django.conf import settings
+from django.contrib.auth.models import AnonymousUser
 from django.core.exceptions import ValidationError
 from django.db import connection
 from django.http import HttpRequest, HttpResponse
@@ -26,18 +27,21 @@ from sqlalchemy.sql import (
     union_all,
 )
 
+from zerver.context_processors import get_realm_from_request
 from zerver.decorator import REQ, has_request_variables
 from zerver.lib.actions import recipient_for_user_profiles
 from zerver.lib.addressee import get_user_profiles, get_user_profiles_by_ids
 from zerver.lib.exceptions import ErrorCode, JsonableError
 from zerver.lib.message import get_first_visible_message_id, messages_for_ids
-from zerver.lib.response import json_error, json_success
+from zerver.lib.narrow import is_web_public_compatible, is_web_public_narrow
+from zerver.lib.response import json_error, json_success, json_unauthorized
 from zerver.lib.sqlalchemy_utils import get_sqlalchemy_connection
 from zerver.lib.streams import (
     can_access_stream_history_by_id,
     can_access_stream_history_by_name,
     get_public_streams_queryset,
     get_stream_by_narrow_operand_access_unchecked,
+    get_web_public_streams_queryset,
 )
 from zerver.lib.topic import DB_TOPIC_NAME, MATCH_TOPIC, topic_column_sa, topic_match_sa
 from zerver.lib.topic_mutes import exclude_topic_mutes
@@ -132,10 +136,12 @@ class NarrowBuilder:
     #  * anything that would pull in additional rows, or information on
     #    other messages.
 
-    def __init__(self, user_profile: UserProfile, msg_id_column: str) -> None:
+    def __init__(self, user_profile: Optional[UserProfile], msg_id_column: str,
+                 realm: Realm, is_web_public_query: bool=False) -> None:
         self.user_profile = user_profile
         self.msg_id_column = msg_id_column
-        self.realm = user_profile.realm
+        self.realm = realm
+        self.is_web_public_query = is_web_public_query
 
     def add_term(self, query: Query, term: Dict[str, Any]) -> Query:
         """
@@ -178,6 +184,10 @@ class NarrowBuilder:
         return query.where(maybe_negate(cond))
 
     def by_in(self, query: Query, operand: str, maybe_negate: ConditionTransform) -> Query:
+        # This operator does not support is_web_public_query.
+        assert not self.is_web_public_query
+        assert self.user_profile is not None
+
         if operand == 'home':
             conditions = exclude_muting_conditions(self.user_profile, [])
             return query.where(and_(*conditions))
@@ -187,6 +197,10 @@ class NarrowBuilder:
         raise BadNarrowOperator("unknown 'in' operand " + operand)
 
     def by_is(self, query: Query, operand: str, maybe_negate: ConditionTransform) -> Query:
+        # This operator class does not support is_web_public_query.
+        assert not self.is_web_public_query
+        assert self.user_profile is not None
+
         if operand == 'private':
             cond = column("flags").op("&")(UserMessage.flags.is_private.mask) != 0
             return query.where(maybe_negate(cond))
@@ -234,6 +248,9 @@ class NarrowBuilder:
             # private streams you are no longer subscribed to, we
             # need get_stream_by_narrow_operand_access_unchecked here.
             stream = get_stream_by_narrow_operand_access_unchecked(operand, self.realm)
+
+            if self.is_web_public_query and not stream.is_web_public:
+                raise BadNarrowOperator('unknown web-public stream ' + str(operand))
         except Stream.DoesNotExist:
             raise BadNarrowOperator('unknown stream ' + str(operand))
 
@@ -269,6 +286,8 @@ class NarrowBuilder:
             # Get all both subscribed and non subscribed public streams
             # but exclude any private subscribed streams.
             recipient_queryset = get_public_streams_queryset(self.realm)
+        elif operand == 'web-public':
+            recipient_queryset = get_web_public_streams_queryset(self.realm)
         else:
             raise BadNarrowOperator('unknown streams operand ' + operand)
 
@@ -344,6 +363,9 @@ class NarrowBuilder:
 
     def by_pm_with(self, query: Query, operand: Union[str, Iterable[int]],
                    maybe_negate: ConditionTransform) -> Query:
+        # This operator does not support is_web_public_query.
+        assert not self.is_web_public_query
+        assert self.user_profile is not None
 
         try:
             if isinstance(operand, str):
@@ -405,6 +427,10 @@ class NarrowBuilder:
 
     def by_group_pm_with(self, query: Query, operand: Union[str, int],
                          maybe_negate: ConditionTransform) -> Query:
+        # This operator does not support is_web_public_query.
+        assert not self.is_web_public_query
+        assert self.user_profile is not None
+
         try:
             if isinstance(operand, str):
                 narrow_profile = get_user_including_cross_realm(operand, self.realm)
@@ -580,7 +606,8 @@ def narrow_parameter(json: str) -> OptionalNarrowListT:
 
     return list(map(convert_term, data))
 
-def ok_to_include_history(narrow: OptionalNarrowListT, user_profile: UserProfile) -> bool:
+def ok_to_include_history(narrow: OptionalNarrowListT, user_profile: Optional[UserProfile],
+                          is_web_public_query: bool) -> bool:
     # There are occasions where we need to find Message rows that
     # have no corresponding UserMessage row, because the user is
     # reading a public stream that might include messages that
@@ -591,6 +618,17 @@ def ok_to_include_history(narrow: OptionalNarrowListT, user_profile: UserProfile
     # query that narrows to a particular public stream on the user's realm.
     # If we screw this up, then we can get into a nasty situation of
     # polluting our narrow results with messages from other realms.
+
+    # For web-public queries, we are always returning history.  The
+    # analogues of the below stream access checks for whether streams
+    # have is_web_public set and banning is operators in this code
+    # path are done directly in NarrowBuilder.
+    if is_web_public_query:
+        assert user_profile is None
+        return True
+
+    assert user_profile is not None
+
     include_history = False
     if narrow is not None:
         for term in narrow:
@@ -650,7 +688,7 @@ def exclude_muting_conditions(user_profile: UserProfile,
 
     return conditions
 
-def get_base_query_for_search(user_profile: UserProfile,
+def get_base_query_for_search(user_profile: Optional[UserProfile],
                               need_message: bool,
                               need_user_message: bool) -> Tuple[Query, ColumnElement]:
     # Handle the simple case where user_message isn't involved first.
@@ -662,6 +700,7 @@ def get_base_query_for_search(user_profile: UserProfile,
         inner_msg_id_col = literal_column("zerver_message.id")
         return (query, inner_msg_id_col)
 
+    assert user_profile is not None
     if need_message:
         query = select([column("message_id"), column("flags")],
                        column("user_profile_id") == literal(user_profile.id),
@@ -677,17 +716,19 @@ def get_base_query_for_search(user_profile: UserProfile,
     inner_msg_id_col = column("message_id")
     return (query, inner_msg_id_col)
 
-def add_narrow_conditions(user_profile: UserProfile,
+def add_narrow_conditions(user_profile: Optional[UserProfile],
                           inner_msg_id_col: ColumnElement,
                           query: Query,
-                          narrow: OptionalNarrowListT) -> Tuple[Query, bool]:
+                          narrow: OptionalNarrowListT,
+                          is_web_public_query: bool,
+                          realm: Realm) -> Tuple[Query, bool]:
     is_search = False  # for now
 
     if narrow is None:
         return (query, is_search)
 
     # Build the query for the narrow
-    builder = NarrowBuilder(user_profile, inner_msg_id_col)
+    builder = NarrowBuilder(user_profile, inner_msg_id_col, realm, is_web_public_query)
     search_operands = []
 
     # As we loop through terms, builder does most of the work to extend
@@ -711,8 +752,13 @@ def add_narrow_conditions(user_profile: UserProfile,
     return (query, is_search)
 
 def find_first_unread_anchor(sa_conn: Any,
-                             user_profile: UserProfile,
+                             user_profile: Optional[UserProfile],
                              narrow: OptionalNarrowListT) -> int:
+    # For anonymous web users, all messages are treated as read, and so
+    # always return LARGER_THAN_MAX_MESSAGE_ID.
+    if user_profile is None:
+        return LARGER_THAN_MAX_MESSAGE_ID
+
     # We always need UserMessage in our query, because it has the unread
     # flag for the user.
     need_user_message = True
@@ -735,6 +781,8 @@ def find_first_unread_anchor(sa_conn: Any,
         inner_msg_id_col=inner_msg_id_col,
         query=query,
         narrow=narrow,
+        is_web_public_query=False,
+        realm=user_profile.realm,
     )
 
     condition = column("flags").op("&")(UserMessage.flags.read.mask) == 0
@@ -792,7 +840,8 @@ def parse_anchor_value(anchor_val: Optional[str],
         raise JsonableError(_("Invalid anchor"))
 
 @has_request_variables
-def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
+def get_messages_backend(request: HttpRequest,
+                         maybe_user_profile: Union[UserProfile, AnonymousUser],
                          anchor_val: Optional[str]=REQ(
                              'anchor', str_validator=check_string, default=None),
                          num_before: int=REQ(converter=to_non_negative_int),
@@ -808,19 +857,54 @@ def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
             MAX_MESSAGES_PER_FETCH,
         ))
 
-    if user_profile.realm.email_address_visibility != Realm.EMAIL_ADDRESS_VISIBILITY_EVERYONE:
+    if not maybe_user_profile.is_authenticated:
+        # If user is not authenticated, clients must include
+        # `streams:web-public` in their narrow query to indicate this
+        # is a web-public query.  This helps differentiate between
+        # cases of web-public queries (where we should return the
+        # web-public results only) and clients with buggy
+        # authentication code (where we should return an auth error).
+        if not is_web_public_narrow(narrow):
+            return json_unauthorized()
+        assert narrow is not None
+        if not is_web_public_compatible(narrow):
+            return json_unauthorized()
+
+        realm = get_realm_from_request(request)
+        if realm is None:
+            return json_error(_("Invalid subdomain."))
+
+        # We use None to indicate unauthenticated requests as it's more
+        # readable than using AnonymousUser, and the lack of Django
+        # stubs means that mypy can't check AnonymousUser well.
+        user_profile: Optional[UserProfile] = None
+        is_web_public_query = True
+    else:
+        assert isinstance(maybe_user_profile, UserProfile)
+        user_profile = maybe_user_profile
+        assert user_profile is not None
+        realm = user_profile.realm
+        is_web_public_query = False
+
+    assert realm is not None
+
+    if is_web_public_query or \
+            realm.email_address_visibility != Realm.EMAIL_ADDRESS_VISIBILITY_EVERYONE:
         # If email addresses are only available to administrators,
         # clients cannot compute gravatars, so we force-set it to false.
         client_gravatar = False
 
-    include_history = ok_to_include_history(narrow, user_profile)
+    include_history = ok_to_include_history(narrow, user_profile, is_web_public_query)
     if include_history:
         # The initial query in this case doesn't use `zerver_usermessage`,
         # and isn't yet limited to messages the user is entitled to see!
         #
         # This is OK only because we've made sure this is a narrow that
-        # will cause us to limit the query appropriately later.
+        # will cause us to limit the query appropriately elsewhere.
         # See `ok_to_include_history` for details.
+        #
+        # Note that is_web_public_query=True goes here, since
+        # include_history is semantically correct for is_web_public_query.
         need_message = True
         need_user_message = False
     elif narrow is None:
@@ -843,6 +927,8 @@ def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
         inner_msg_id_col=inner_msg_id_col,
         query=query,
         narrow=narrow,
+        realm=realm,
+        is_web_public_query=is_web_public_query,
     )
 
     if narrow is not None:
@@ -858,7 +944,7 @@ def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
     sa_conn = get_sqlalchemy_connection()
 
     if anchor is None:
-        # The use_first_unread_anchor code path
+        # `anchor=None` corresponds to the anchor="first_unread" parameter.
         anchor = find_first_unread_anchor(
             sa_conn,
             user_profile,
@@ -873,7 +959,8 @@ def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
     if anchored_to_right:
         num_after = 0
 
-    first_visible_message_id = get_first_visible_message_id(user_profile.realm)
+    first_visible_message_id = get_first_visible_message_id(realm)
+
     query = limit_query_to_range(
         query=query,
         num_before=num_before,
@@ -912,7 +999,14 @@ def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
     # 'messages' list.
     message_ids: List[int] = []
     user_message_flags: Dict[int, List[str]] = {}
-    if include_history:
+    if is_web_public_query:
+        # For web-public users, we treat all historical messages as read.
+        for row in rows:
+            message_id = row[0]
+            message_ids.append(message_id)
+            user_message_flags[message_id] = ["read"]
+    elif include_history:
+        assert user_profile is not None
         message_ids = [row[0] for row in rows]
 
         # TODO: This could be done with an outer join instead of two queries
@@ -951,7 +1045,7 @@ def get_messages_backend(request: HttpRequest, user_profile: UserProfile,
         search_fields=search_fields,
         apply_markdown=apply_markdown,
         client_gravatar=client_gravatar,
-        allow_edit_history=user_profile.realm.allow_edit_history,
+        allow_edit_history=realm.allow_edit_history,
     )
 
     statsd.incr('loaded_old_messages', len(message_list))
@@ -1111,6 +1205,7 @@ def post_process_limited_query(rows: List[Any],
         found_oldest=found_oldest,
         history_limited=history_limited,
     )
+
 @has_request_variables
 def messages_in_narrow_backend(request: HttpRequest, user_profile: UserProfile,
                                msg_ids: List[int]=REQ(validator=check_list(check_int)),
@@ -1128,7 +1223,7 @@ def messages_in_narrow_backend(request: HttpRequest, user_profile: UserProfile,
                         literal_column("zerver_usermessage.message_id") ==
                         literal_column("zerver_message.id")))
 
-    builder = NarrowBuilder(user_profile, column("message_id"))
+    builder = NarrowBuilder(user_profile, column("message_id"), user_profile.realm)
     if narrow is not None:
         for term in narrow:
             query = builder.add_term(query, term)

zproject/urls.py

@@ -208,7 +208,8 @@ v1_api_and_json_patterns = [
     # messages -> zerver.views.message*
     # GET returns messages, possibly filtered, POST sends a message
     path('messages', rest_dispatch,
-         {'GET': 'zerver.views.message_fetch.get_messages_backend',
+         {'GET': ('zerver.views.message_fetch.get_messages_backend',
+                  {'allow_anonymous_user_web'}),
           'POST': ('zerver.views.message_send.send_message_backend',
                    {'allow_incoming_webhooks'})}),
     path('messages/<int:message_id>', rest_dispatch,