Add more validation and logging to policy server signature responses.

changelog.d/19807.misc

@@ -0,0 +1 @@
+Add more validation and logging to policy server signature responses.

synapse/api/errors.py

@@ -806,7 +806,18 @@ def __init__(
 
 class HttpResponseException(CodeMessageException):
     """
-    Represents an HTTP-level failure of an outbound request
+    Represents an HTTP-level failure of an outbound request,
+    where the response has an unexpected status code.
+
+    The response may contain a JSON-encoded Matrix API error
+    with an `errcode` and `error`, but this is optional.
+
+    Analogous to `InvalidResponseError`, but at the response code level
+    rather than the response body level.
+
+    Should not be allowed to bubble to an API response without handling.
+    If it does, it will yield a 500 Internal Server Error as any other
+    unhandled exception.
 
     Attributes:
         response: body of response
@@ -824,7 +835,7 @@ def __init__(self, code: int, msg: str, response: bytes):
         self.response = response
 
     def to_synapse_error(self) -> SynapseError:
-        """Make a SynapseError based on an HTTPResponseException
+        """Make a SynapseError based on an HttpResponseException
 
         This is useful when a proxied request has failed, and we need to
         decide how to map the failure onto a matrix error to send back to the
@@ -856,6 +867,50 @@ def to_synapse_error(self) -> SynapseError:
         return ProxiedRequestError(self.code, errmsg, errcode, j)
 
 
+class InvalidResponseError(RuntimeError):
+    """
+    Represents a failure to parse/validate the body returned
+    by an outbound request.
+    Analogous to `HttpResponseException`, but at the response body level
+    rather than the response code level.
+
+    Like `HttpResponseException`, should not be allowed to bubble to
+    an API response without handling.
+    If it does, it will yield a 500 Internal Server Error as any other
+    unhandled exception.
+
+    Attributes:
+        response: body of response
+    """
+
+    def __init__(self, msg: str):
+        """
+        Args:
+            msg: A message for logging purposes.
+        """
+        super().__init__(msg)
+
+    def to_synapse_error(self) -> SynapseError:
+        """Make a SynapseError based on this error.
+
+        The errcode is set to M_UNKNOWN
+        and the error message is set to a generic message.
+        The detail message of this error is not exposed.
+
+        (Maybe we could consider exposing the violating server's name in
+        the error message.)
+
+        Returns:
+            The error converted to a SynapseError.
+        """
+
+        return ProxiedRequestError(
+            HTTPStatus.BAD_GATEWAY,
+            "Remote server presented an invalid response over federation.",
+            Codes.UNKNOWN,
+        )
+
+
 class HomeServerNotSetupException(Exception):
     """
     Raised when an operation is attempted on the HomeServer before setup() has been called.

synapse/federation/federation_client.py

@@ -19,8 +19,6 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-
-
 import copy
 import itertools
 import logging
@@ -37,17 +35,20 @@
     Optional,
     Sequence,
     TypeVar,
+    overload,
 )
 
 import attr
 from prometheus_client import Counter
+from pydantic import ValidationError
 
 from synapse.api.constants import Direction, EventContentFields, EventTypes, Membership
 from synapse.api.errors import (
     CodeMessageException,
     Codes,
     FederationDeniedError,
     HttpResponseException,
+    InvalidResponseError,
     RequestSendFailed,
     SynapseError,
     UnsupportedRoomVersionError,
@@ -72,9 +73,11 @@
 from synapse.logging.opentracing import SynapseTags, log_kv, set_tag, tag_args, trace
 from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import JsonDict, StrCollection, UserID, get_domain_from_id
+from synapse.types.federation.policy import PolicySignResponse
 from synapse.util.async_helpers import concurrently_execute
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.duration import Duration
+from synapse.util.pydantic_models import ParseModel, StrictRootModel
 from synapse.util.retryutils import NotRetryingDestination
 
 if TYPE_CHECKING:
@@ -104,12 +107,6 @@ class PulledPduInfo:
     pull_origin: str
 
 
-class InvalidResponseError(RuntimeError):
-    """Helper for _try_destination_list: indicates that the server returned a response
-    we couldn't parse
-    """
-
-
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class SendJoinResult:
     # The event to persist.
@@ -127,6 +124,41 @@ class SendJoinResult:
     servers_in_room: AbstractSet[str]
 
 
+MODEL_ROOT = TypeVar("MODEL_ROOT", bound=StrictRootModel)
+MODEL_PARSE = TypeVar("MODEL_PARSE", bound=ParseModel)
+
+
+@overload
+def validate_response(
+    content: JsonDict, model_type: type[MODEL_ROOT]
+) -> MODEL_ROOT: ...
+
+
+@overload
+def validate_response(
+    content: JsonDict, model_type: type[MODEL_PARSE]
+) -> MODEL_PARSE: ...
+
+
+# note: this signature is supposed to be ignored by the overload,
+# but yet required, with `no-untyped-def` error given if omitted
+def validate_response(
+    content: JsonDict, model_type: type[MODEL_ROOT] | type[MODEL_PARSE]
+) -> MODEL_ROOT | MODEL_PARSE:
+    """Validate a deserialized JSON object using the given pydantic model.
+
+    Raises:
+        SynapseError if the request body couldn't be decoded as JSON or
+            if it wasn't a JSON object.
+    """
+    try:
+        instance = model_type.model_validate(content)
+    except ValidationError as e:
+        raise InvalidResponseError(str(e))
+
+    return instance
+
+
 class FederationClient(FederationBase):
     def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
@@ -441,7 +473,7 @@ async def _record_failure_callback(
     @tag_args
     async def ask_policy_server_to_sign_event(
         self, destination: str, pdu: EventBase, timeout: int | None = None
-    ) -> JsonDict:
+    ) -> PolicySignResponse:
         """Requests that the destination server (typically a policy server)
         sign the event as not spam.
 
@@ -462,9 +494,10 @@ async def ask_policy_server_to_sign_event(
             pdu.event_id,
             destination,
         )
-        return await self.transport_layer.ask_policy_server_to_sign_event(
+        json_response = await self.transport_layer.ask_policy_server_to_sign_event(
             destination, pdu, timeout=timeout
         )
+        return validate_response(json_response, PolicySignResponse)
 
     @trace
     @tag_args

synapse/handlers/room_policy.py

@@ -22,7 +22,12 @@
 from unpaddedbase64 import decode_base64
 
 from synapse.api.constants import EventTypes
-from synapse.api.errors import Codes, HttpResponseException, SynapseError
+from synapse.api.errors import (
+    Codes,
+    HttpResponseException,
+    InvalidResponseError,
+    SynapseError,
+)
 from synapse.crypto.keyring import VerifyJsonRequest
 from synapse.events import EventBase
 from synapse.util.stringutils import parse_and_validate_server_name
@@ -225,7 +230,7 @@ async def ask_policy_server_to_sign_event(
 
         # Ask the policy server to sign this event.
         try:
-            signature = await self._federation_client.ask_policy_server_to_sign_event(
+            sign_response = await self._federation_client.ask_policy_server_to_sign_event(
                 policy_server.server_name,
                 event,
                 # We set a smallish timeout here as we don't want to block event sending
@@ -241,7 +246,7 @@ async def ask_policy_server_to_sign_event(
             # also be implementation bugs. Whoever reads this when removing unstable MSC4284
             # stuff, make a decision on whether to remove this bit.
             # https://github.com/element-hq/synapse/issues/19502
-            if not signature or len(signature) == 0:
+            if len(sign_response.root) == 0:
                 raise SynapseError(
                     403,
                     "This event has been rejected as probable spam by the policy server",
@@ -261,10 +266,22 @@ async def ask_policy_server_to_sign_event(
             # servers need to manually fetch signatures for. This is the code that allows
             # those events to continue working (because they're legally sent, even if missing
             # the policy server signature).
-            signatures = signature.get(policy_server.server_name, {})
-            for key_id, sig in signatures.items():
-                event.signatures.add_signature(policy_server.server_name, key_id, sig)
-        except HttpResponseException as ex:
+            for key_id, signature_b64 in sign_response.root.items():
+                if (
+                    event.signatures.get_signature(policy_server.server_name, key_id)
+                    is None
+                ):
+                    event.signatures.add_signature(
+                        policy_server.server_name, key_id, signature_b64
+                    )
+                else:
+                    logger.warning(
+                        "Policy server %r attempted to overwrite existing signature by key_id = %r on event %s, ignoring",
+                        policy_server.server_name,
+                        key_id,
+                        event.event_id,
+                    )
+        except (HttpResponseException, InvalidResponseError) as ex:
             # re-wrap HTTP errors as `SynapseError` so they can be proxied to clients directly
             raise ex.to_synapse_error() from ex
 

synapse/types/federation/policy.py

@@ -0,0 +1,51 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl_3.0.html>.
+#
+from pydantic import model_validator
+from typing_extensions import Self
+
+from synapse.util.pydantic_models import StrictRootModel
+
+
+class PolicySignResponse(StrictRootModel[dict[str, str]]):
+    """
+    Response to `POST /_matrix/policy/v1/sign`
+    Spec: https://spec.matrix.org/v1.18/server-server-api/#post_matrixpolicyv1sign
+
+    Example:
+        {
+          "policy.example.org": {
+            "ed25519:policy_server": "zLFxllD0pbBuBpfHh8NuHNaICpReF/PAOpUQTsw+bFGKiGfDNAsnhcP7pbrmhhpfbOAxIdLraQLeeiXBryLmBw"
+          }
+        }
+    """
+
+    @model_validator(mode="after")
+    def check_rules(self) -> Self:
+        """
+        > The Policy Server has signed the event, indicating that it recommends the event for inclusion in the room.
+        > Only the Policy Server’s signature is returned.
+        > This signature is to be added to the event before sending or processing the event further.
+        >
+        > `ed25519:policy_server` is always used for Ed25519 signatures.
+        > — https://spec.matrix.org/v1.18/server-server-api/#validating-policy-server-signatures
+        """
+
+        for key_id in self.root.keys():
+            if key_id.startswith("ed25519:") and key_id != "ed25519:policy_server":
+                # > `ed25519:policy_server` is always used for Ed25519 signatures.
+                raise ValueError(
+                    "policy servers must only use ed25519:policy_server for Ed25519 signatures"
+                )
+
+        return self

synapse/util/pydantic_models.py

@@ -13,9 +13,16 @@
 #
 #
 
-from typing import Annotated
+from typing import Annotated, TypeVar
 
-from pydantic import AfterValidator, BaseModel, ConfigDict, StrictStr, StringConstraints
+from pydantic import (
+    AfterValidator,
+    BaseModel,
+    ConfigDict,
+    RootModel,
+    StrictStr,
+    StringConstraints,
+)
 
 from synapse.api.errors import SynapseError
 from synapse.types import EventID
@@ -42,12 +49,25 @@ class ParseModel(BaseModel):
     server operators.
 
     Subclassing in this way is recommended by
-    https://pydantic-docs.helpmanual.io/usage/model_config/#change-behaviour-globally
+    https://pydantic.dev/docs/validation/latest/concepts/config/#change-behaviour-globally
+    [accessed at 2026-05-26]
     """
 
     model_config = ConfigDict(extra="ignore", frozen=True, strict=True)
 
 
+T = TypeVar("T")
+
+
+class StrictRootModel(RootModel[T]):
+    """
+    A custom version of Pydantic's RootModel, in the same vein as `ParseModel`.
+    Refer to `ParseModel` above for the configuration details.
+    """
+
+    model_config = ConfigDict(frozen=True, strict=True)
+
+
 def validate_event_id_v1_and_2(value: str) -> str:
     try:
         EventID.from_string(value)