Feature: return errors for invalid message API queries (#335)

Problem: the messages.json endpoint was too permissive, allowing users to specify invalid chains, message types, make incoherent pagination / time filtering, etc.

Solution: detect these cases and return a 422 error.

Replaced the custom validation code of the endpoint by a Pydantic model. Added tests for start/end date parameters and pagination.

Breaking changes:
* The "endDate" field is now considered as exclusive. Only messages with a time field strictly lower than the value will be returned, instead of a value lower or equal before.
* The endpoint now returns a 422 error instead of a 400 when incorrect parameters are passed.

Moreover, a 422 error code will now be returned in the following situations, where the previous implementation would simply return a 200:
* if an invalid/unknown chain appears in the "chains".
* if an invalid/unknown message type is specified in "msgType".
* if an invalid item hash (=not a hexadecimal sha256, CIDv0 or CIDv1) is specified in the "hashes" or "contentHashes" field.
* if the "endDate" field is lower than the "startDate" field.
* if "endDate" or "startDate" are negative.
* if pagination parameters ("page" and "pagination") are negative.

Author: odesenfans

src/aleph/web/controllers/messages.py

@@ -1,129 +1,200 @@
-from typing import Any, Dict, List, Optional, Set
+import asyncio
+import logging
+from enum import IntEnum
+from typing import Any, Dict, List, Optional, Mapping
 
-from aleph.model.messages import CappedMessage, Message
 from aiohttp import web
-from aiohttp.web_exceptions import HTTPBadRequest
-import asyncio
-from pymongo.cursor import CursorType
+from aleph_message.models import MessageType, ItemHash, Chain
 from bson.objectid import ObjectId
-from aleph.web.controllers.utils import Pagination, cond_output, prepare_date_filters
-import logging
+from pydantic import BaseModel, Field, validator, ValidationError, root_validator
+from pymongo.cursor import CursorType
 
-LOGGER = logging.getLogger("MESSAGES")
-
-KNOWN_QUERY_FIELDS = {
-    "sort_order",
-    "msgType",
-    "addresses",
-    "refs",
-    "contentHashes",
-    "contentKeys",
-    "contentTypes",
-    "chains",
-    "channels",
-    "tags",
-    "hashes",
-    "history",
-    "pagination",
-    "page",  # page is handled in Pagination.get_pagination_params
-    "startDate",
-    "endDate",
-}
-
-
-async def get_filters(request: web.Request):
-    def get_query_list_field(field: str, separator=",") -> Optional[List[str]]:
-        field_str = request.query.get(field, None)
-        return field_str.split(separator) if field_str is not None else None
-
-    unknown_query_fields: Set[str] = set(request.query.keys()).difference(
-        KNOWN_QUERY_FIELDS
-    )
-    if unknown_query_fields:
-        raise ValueError(f"Unknown query fields: {unknown_query_fields}")
-
-    find_filters: Dict[str, Any] = {}
-
-    msg_type = request.query.get("msgType", None)
-
-    filters: List[Dict[str, Any]] = []
-    addresses = get_query_list_field("addresses")
-    refs = get_query_list_field("refs")
-    content_keys = get_query_list_field("contentKeys")
-    content_hashes = get_query_list_field("contentHashes")
-    content_types = get_query_list_field("contentTypes")
-    chains = get_query_list_field("chains")
-    channels = get_query_list_field("channels")
-    tags = get_query_list_field("tags")
-    hashes = get_query_list_field("hashes")
-
-    date_filters = prepare_date_filters(request, "time")
-
-    if msg_type is not None:
-        filters.append({"type": msg_type})
-
-    if addresses is not None:
-        filters.append(
-            {
-                "$or": [
-                    {"content.address": {"$in": addresses}},
-                    {"sender": {"$in": addresses}},
-                ]
-            }
-        )
+from aleph.model.messages import CappedMessage, Message
+from aleph.web.controllers.utils import (
+    LIST_FIELD_SEPARATOR,
+    Pagination,
+    cond_output,
+    make_date_filters,
+)
 
-    if content_hashes is not None:
-        filters.append({"content.item_hash": {"$in": content_hashes}})
+LOGGER = logging.getLogger(__name__)
 
-    if content_keys is not None:
-        filters.append({"content.key": {"$in": content_keys}})
 
-    if content_types is not None:
-        filters.append({"content.type": {"$in": content_types}})
+DEFAULT_MESSAGES_PER_PAGE = 20
+DEFAULT_PAGE = 1
+DEFAULT_WS_HISTORY = 10
 
-    if refs is not None:
-        filters.append({"content.ref": {"$in": refs}})
 
-    if tags is not None:
-        filters.append({"content.content.tags": {"$elemMatch": {"$in": tags}}})
+class SortOrder(IntEnum):
+    ASCENDING = 1
+    DESCENDING = -1
 
-    if chains is not None:
-        filters.append({"chain": {"$in": chains}})
 
-    if channels is not None:
-        filters.append({"channel": {"$in": channels}})
+class MessageQueryParams(BaseModel):
+    sort_order: SortOrder = Field(
+        default=SortOrder.DESCENDING,
+        description="Order in which messages should be listed: "
+        "-1 means most recent messages first, 1 means older messages first.",
+    )
+    message_type: Optional[MessageType] = Field(
+        default=None, alias="msgType", description="Message type."
+    )
+    addresses: Optional[List[str]] = Field(
+        default=None, description="Accepted values for the 'sender' field."
+    )
+    refs: Optional[List[str]] = Field(
+        default=None, description="Accepted values for the 'content.ref' field."
+    )
+    content_hashes: Optional[List[ItemHash]] = Field(
+        default=None,
+        alias="contentHashes",
+        description="Accepted values for the 'content.item_hash' field.",
+    )
+    content_keys: Optional[List[ItemHash]] = Field(
+        default=None,
+        alias="contentKeys",
+        description="Accepted values for the 'content.keys' field.",
+    )
+    content_types: Optional[List[ItemHash]] = Field(
+        default=None,
+        alias="contentTypes",
+        description="Accepted values for the 'content.type' field.",
+    )
+    chains: Optional[List[Chain]] = Field(
+        default=None, description="Accepted values for the 'chain' field."
+    )
+    channels: Optional[List[str]] = Field(
+        default=None, description="Accepted values for the 'channel' field."
+    )
+    tags: Optional[List[str]] = Field(
+        default=None, description="Accepted values for the 'content.content.tag' field."
+    )
+    hashes: Optional[List[ItemHash]] = Field(
+        default=None, description="Accepted values for the 'item_hash' field."
+    )
+    history: Optional[int] = Field(
+        DEFAULT_WS_HISTORY,
+        ge=10,
+        lt=200,
+        description="Accepted values for the 'item_hash' field.",
+    )
+    pagination: int = Field(
+        default=DEFAULT_MESSAGES_PER_PAGE,
+        ge=0,
+        description="Maximum number of messages to return. Specifying 0 removes this limit.",
+    )
+    page: int = Field(
+        default=DEFAULT_PAGE, ge=1, description="Offset in pages. Starts at 1."
+    )
+    start_date: float = Field(
+        default=0,
+        ge=0,
+        alias="startDate",
+        description="Start date timestamp. If specified, only messages with "
+        "a time field greater or equal to this value will be returned.",
+    )
+    end_date: float = Field(
+        default=0,
+        ge=0,
+        alias="endDate",
+        description="End date timestamp. If specified, only messages with "
+        "a time field lower than this value will be returned.",
+    )
 
-    if hashes is not None:
-        filters.append(
-            {"$or": [{"item_hash": {"$in": hashes}}, {"tx_hash": {"$in": hashes}}]}
-        )
+    @root_validator
+    def validate_field_dependencies(cls, values):
+        start_date = values.get("start_date")
+        end_date = values.get("end_date")
+        if start_date and end_date and (end_date < start_date):
+            raise ValueError("end date cannot be lower than start date.")
+        return values
+
+    @validator(
+        "addresses",
+        "content_hashes",
+        "content_keys",
+        "content_types",
+        "chains",
+        "channels",
+        "tags",
+        pre=True,
+    )
+    def split_str(cls, v):
+        if isinstance(v, str):
+            return v.split(LIST_FIELD_SEPARATOR)
+        return v
+
+    def to_mongodb_filters(self) -> Mapping[str, Any]:
+        filters: List[Dict[str, Any]] = []
+
+        if self.message_type is not None:
+            filters.append({"type": self.message_type})
+
+        if self.addresses is not None:
+            filters.append(
+                {
+                    "$or": [
+                        {"content.address": {"$in": self.addresses}},
+                        {"sender": {"$in": self.addresses}},
+                    ]
+                }
+            )
 
-    if date_filters is not None:
-        filters.append(date_filters)
+        if self.content_hashes is not None:
+            filters.append({"content.item_hash": {"$in": self.content_hashes}})
+        if self.content_keys is not None:
+            filters.append({"content.key": {"$in": self.content_keys}})
+        if self.content_types is not None:
+            filters.append({"content.type": {"$in": self.content_types}})
+        if self.refs is not None:
+            filters.append({"content.ref": {"$in": self.refs}})
+        if self.tags is not None:
+            filters.append({"content.content.tags": {"$elemMatch": {"$in": self.tags}}})
+        if self.chains is not None:
+            filters.append({"chain": {"$in": self.chains}})
+        if self.channels is not None:
+            filters.append({"channel": {"$in": self.channels}})
+        if self.hashes is not None:
+            filters.append(
+                {
+                    "$or": [
+                        {"item_hash": {"$in": self.hashes}},
+                        {"tx_hash": {"$in": self.hashes}},
+                    ]
+                }
+            )
+
+        date_filters = make_date_filters(
+            start=self.start_date, end=self.end_date, filter_key="time"
+        )
+        if date_filters:
+            filters.append(date_filters)
 
-    if len(filters) > 0:
-        find_filters = {"$and": filters} if len(filters) > 1 else filters[0]
+        and_filter = {}
+        if filters:
+            and_filter = {"$and": filters} if len(filters) > 1 else filters[0]
 
-    return find_filters
+        return and_filter
 
 
 async def view_messages_list(request):
     """Messages list view with filters"""
 
     try:
-        find_filters = await get_filters(request)
-    except ValueError as error:
-        raise HTTPBadRequest(body=error.args[0])
-
-    (
-        pagination_page,
-        pagination_per_page,
-        pagination_skip,
-    ) = Pagination.get_pagination_params(request)
-    if pagination_per_page is None:
-        pagination_per_page = 0
-    if pagination_skip is None:
-        pagination_skip = 0
+        query_params = MessageQueryParams.parse_obj(request.query)
+    except ValidationError as e:
+        raise web.HTTPUnprocessableEntity(body=e.json(indent=4))
+
+    # If called from the messages/page/{page}.json endpoint, override the page
+    # parameters with the URL one
+    if url_page_param := request.match_info.get("page"):
+        query_params.page = int(url_page_param)
+
+    find_filters = query_params.to_mongodb_filters()
+
+    pagination_page = query_params.page
+    pagination_per_page = query_params.pagination
+    pagination_skip = (query_params.page - 1) * query_params.pagination
 
     messages = [
         msg
@@ -132,14 +203,14 @@ async def view_messages_list(request):
             projection={"_id": 0},
             limit=pagination_per_page,
             skip=pagination_skip,
-            sort=[("time", int(request.query.get("sort_order", "-1")))],
+            sort=[("time", query_params.sort_order.value)],
         )
     ]
 
     context = {"messages": messages}
 
     if pagination_per_page is not None:
-        if len(find_filters.keys()):
+        if find_filters:
             total_msgs = await Message.collection.count_documents(find_filters)
         else:
             total_msgs = await Message.collection.estimated_document_count()
@@ -173,11 +244,10 @@ async def messages_ws(request: web.Request):
     collection = CappedMessage.collection
     last_id = None
 
-    find_filters = await get_filters(request)
-    initial_count = int(request.query.get("history", 10))
-    initial_count = max(initial_count, 10)
-    # let's cap this to 200 historic messages max.
-    initial_count = min(initial_count, 200)
+    query_params = MessageQueryParams.parse_obj(request.query)
+    find_filters = query_params.to_mongodb_filters()
+
+    initial_count = query_params.history
 
     items = [
         item

src/aleph/web/controllers/utils.py

@@ -19,7 +19,9 @@ def get_pagination_params(request):
         with_pagination = pagination_param != 0
 
         if pagination_page < 1:
-            raise web.HTTPBadRequest(text=f"Query field 'page' must be ≥ 1, not {pagination_page}")
+            raise web.HTTPBadRequest(
+                text=f"Query field 'page' must be ≥ 1, not {pagination_page}"
+            )
 
         if not with_pagination:
             pagination_per_page = None
@@ -66,6 +68,21 @@ def iter_pages(self, left_edge=2, left_current=2, right_current=5, right_edge=2)
                 last = num
 
 
+def make_date_filters(start: float, end: float, filter_key: str):
+    filters = []
+    if start:
+        filters.append({filter_key: {"$gte": start}})
+    if end:
+        filters.append({filter_key: {"$lt": end}})
+
+    if len(filters) > 1:
+        return {"$and": filters}
+    if filters:
+        return filters[0]
+
+    return None
+
+
 def prepare_date_filters(request, filter_key):
     date_filters = None