From 436a48baeefae6018d95984cb684295fd786ba48 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Wed, 28 Oct 2026 01:57:58 +0100 Subject: [PATCH] Experimental: the fermion arbiter, a commit sequencer for photon rooms Behind experimental_features.fermion_arbiter_enabled (advertised as org.matrix.experimental.fermion.arbiter). Members of a photon room report each MLS commit they make as {arbiter, base, epoch, event_id, epoch_interval_ms} to their own server at POST /_matrix/client/unstable/org.matrix.experimental.fermion.v1/rooms/{roomId}/commits, which answers itself if it is the room's arbiter and otherwise relays over federation (PUT .../unstable/org.matrix.experimental.fermion.v1/rooms/{roomId}/commits). The first commit seen on the current head is canonical; the arbiter signs rulings naming the canonical extremity and the commits that reached it. At epoch_interval_ms 0 the ruling is the response (a sequencer: vanilla MLS); above it rulings are minted that long after the first unruled commit and fetched from GET .../rulings?arbiter=&from=. The arbiter never joins the group or sees a commit's bytes. --- rust/src/config/mod.rs | 1 + rust/src/handlers/versions.rs | 4 + synapse/config/experimental.py | 6 + .../federation/transport/server/__init__.py | 16 ++ .../federation/transport/server/fermion.py | 88 +++++++ synapse/rest/__init__.py | 2 + synapse/rest/client/fermion.py | 235 ++++++++++++++++++ synapse/server.py | 7 + tests/rest/client/test_fermion.py | 171 +++++++++++++ 9 files changed, 530 insertions(+) create mode 100644 synapse/federation/transport/server/fermion.py create mode 100644 synapse/rest/client/fermion.py create mode 100644 tests/rest/client/test_fermion.py diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs index a0072268e0..6500f1b40a 100644 --- a/rust/src/config/mod.rs +++ b/rust/src/config/mod.rs @@ -77,4 +77,5 @@ pub struct ExperimentalConfig { pub msc4446_enabled: bool, pub msc4502_enabled: bool, pub photon_shared_to_device_enabled: bool, + pub fermion_arbiter_enabled: bool, } diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index 58c625acb7..03401964f8 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -285,6 +285,9 @@ pub struct UnstableFeatureMap { /// Photon: shared to-device content with cached recipient sets #[serde(rename = "org.matrix.experimental.photon.shared_to_device")] photon_shared_to_device: bool, + /// Fermion: this server arbitrates the photon rooms its users create + #[serde(rename = "org.matrix.experimental.fermion.arbiter")] + fermion_arbiter: bool, // Whether new rooms will be set to encrypted or not (based on presets). #[serde(rename = "io.element.e2ee_forced.public")] @@ -340,6 +343,7 @@ pub fn synapse_config_to_global_unstable_feature_map( msc4446_enabled: config.experimental.msc4446_enabled, msc4502: config.experimental.msc4502_enabled, photon_shared_to_device: config.experimental.photon_shared_to_device_enabled, + fermion_arbiter: config.experimental.fermion_arbiter_enabled, e2ee_forced_public: config .room .encryption_enabled_by_default_for_room_presets diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index dbe148833a..4b4d7b5111 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -324,3 +324,9 @@ class ExperimentalConfig(Config): self.photon_shared_to_device_enabled: bool = experimental.get( "photon_shared_to_device_enabled", False ) + + # Fermion: this server arbitrates (orders the MLS commits of) the + # photon rooms its users create. + self.fermion_arbiter_enabled: bool = experimental.get( + "fermion_arbiter_enabled", False + ) diff --git a/synapse/federation/transport/server/__init__.py b/synapse/federation/transport/server/__init__.py index 70fc55123b..7279b84db5 100644 --- a/synapse/federation/transport/server/__init__.py +++ b/synapse/federation/transport/server/__init__.py @@ -342,5 +342,21 @@ def register_servlets( server_name=hs.hostname, ).register(resource) + if ( + "federation" in servlet_groups + and hs.config.experimental.fermion_arbiter_enabled + ): + from synapse.federation.transport.server.fermion import ( + FERMION_SERVLET_CLASSES, + ) + + for servletclass in FERMION_SERVLET_CLASSES: + servletclass( + hs=hs, + authenticator=authenticator, + ratelimiter=ratelimiter, + server_name=hs.hostname, + ).register(resource) + if "federation" in servlet_groups: appservice_proxy.register_servlets(hs, resource, authenticator, ratelimiter) diff --git a/synapse/federation/transport/server/fermion.py b/synapse/federation/transport/server/fermion.py new file mode 100644 index 0000000000..17c78a537e --- /dev/null +++ b/synapse/federation/transport/server/fermion.py @@ -0,0 +1,88 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 New Vector, 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: +# . +# + +"""The fermion arbiter over federation: a member's homeserver relays its +submissions here (see `synapse.rest.client.fermion`).""" + +from typing import TYPE_CHECKING, Literal + +from synapse.api.urls import FEDERATION_UNSTABLE_PREFIX +from synapse.federation.transport.server._base import ( + Authenticator, + BaseFederationServlet, +) +from synapse.http.servlet import parse_integer_from_args +from synapse.types import JsonDict + +if TYPE_CHECKING: + from synapse.server import HomeServer + +PATH = "/org.matrix.experimental.fermion.v1/rooms/(?P[^/]*)" + + +class FermionCommitsServlet(BaseFederationServlet): + PREFIX = FEDERATION_UNSTABLE_PREFIX + PATH = PATH + "/commits" + CATEGORY = "Federation requests" + + def __init__( + self, + hs: "HomeServer", + authenticator: Authenticator, + ratelimiter: object, + server_name: str, + ): + super().__init__(hs, authenticator, ratelimiter, server_name) # type: ignore[arg-type] + self.arbiter = hs.get_fermion_arbiter() + + async def on_PUT( + self, + origin: str, + content: JsonDict, + query: dict[bytes, list[bytes]], + room_id: str, + ) -> tuple[int, JsonDict]: + return 200, self.arbiter.submit(room_id, content, origin) + + +class FermionRulingsServlet(BaseFederationServlet): + PREFIX = FEDERATION_UNSTABLE_PREFIX + PATH = PATH + "/rulings" + CATEGORY = "Federation requests" + + def __init__( + self, + hs: "HomeServer", + authenticator: Authenticator, + ratelimiter: object, + server_name: str, + ): + super().__init__(hs, authenticator, ratelimiter, server_name) # type: ignore[arg-type] + self.arbiter = hs.get_fermion_arbiter() + + async def on_GET( + self, + origin: str, + content: Literal[None], + query: dict[bytes, list[bytes]], + room_id: str, + ) -> tuple[int, JsonDict]: + since = parse_integer_from_args(query, "from", default=0) + return 200, {"rulings": self.arbiter.rulings(room_id, since)} + + +FERMION_SERVLET_CLASSES: tuple[type[BaseFederationServlet], ...] = ( + FermionCommitsServlet, + FermionRulingsServlet, +) diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py index 95400790b1..b7179cd69c 100644 --- a/synapse/rest/__init__.py +++ b/synapse/rest/__init__.py @@ -64,6 +64,7 @@ from synapse.rest.client import ( room_keys, room_membership, room_upgrade_rest_servlet, + fermion, sendtodevice, sync, tags, @@ -117,6 +118,7 @@ CLIENT_SERVLET_FUNCTIONS: tuple[RegisterServletsFunc, ...] = ( devices.register_servlets, thirdparty.register_servlets, sendtodevice.register_servlets, + fermion.register_servlets, user_directory.register_servlets, room_upgrade_rest_servlet.register_servlets, capabilities.register_servlets, diff --git a/synapse/rest/client/fermion.py b/synapse/rest/client/fermion.py new file mode 100644 index 0000000000..3d147e8221 --- /dev/null +++ b/synapse/rest/client/fermion.py @@ -0,0 +1,235 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 New Vector, 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: +# . +# + +"""The fermion arbiter: ERA's finality node for a photon room created on +this server (`docs/boson/PLAN.md` in matrix-rust-sdk). + +The server never joins the room's MLS group and is never sent a commit; it +sees only epoch ids and event ids. Members report each commit they make +(`base`, the DMLS epoch it is built on, and `epoch`, the one it starts), +the server keeps the first commit it saw on its current head as canonical, +and it signs rulings naming the canonical extremity and the commits that +reached it. At `epoch_interval_ms` 0 the ruling comes back in the +response, so a committer can ask before fanning out and drop a losing +commit (vanilla MLS with a sequencer); above 0 rulings are minted that +long after the first unruled commit, and members merge their losing +branches onto the canonical one. + +A member talks to its own homeserver (the client endpoints here, with its +access token), which answers itself if it is the room's arbiter and +otherwise relays over federation (the servlets in +`synapse.federation.transport.server.fermion`); so the arbiter knows the +server and user behind every submission, and a browser never needs to +reach a foreign server. + +ponytail: in-memory, per room, bounded by an LRU; a restart forgets the +head and the next submission re-seeds it. +""" + +import logging +from typing import TYPE_CHECKING + +from signedjson.sign import sign_json + +from synapse.api.errors import SynapseError +from synapse.api.urls import FEDERATION_UNSTABLE_PREFIX +from synapse.http import servlet +from synapse.http.server import HttpServer +from synapse.http.servlet import ( + assert_params_in_dict, + parse_integer, + parse_json_object_from_request, + parse_string, +) +from synapse.http.site import SynapseRequest +from synapse.types import JsonDict +from synapse.util.caches.lrucache import LruCache +from synapse.util.duration import Duration + +from ._base import client_patterns + +if TYPE_CHECKING: + from synapse.server import HomeServer + +logger = logging.getLogger(__name__) + +PREFIX = "/org.matrix.experimental.fermion.v1/rooms/(?P[^/]*)" +FEDERATION_PATH = ( + FEDERATION_UNSTABLE_PREFIX + "/org.matrix.experimental.fermion.v1/rooms/" +) + +# Rulings kept per room for `GET …/rulings?from=`. +RULINGS_KEPT = 64 +MAX_ROOMS = 10_000 + + +class ArbiterRoom: + """What the arbiter remembers of one room.""" + + def __init__(self) -> None: + # The canonical extremity: the DMLS epoch id commits must build on. + # None until the first submission seeds it with its base. + self.head: str | None = None + # Event ids of the commits accepted since the last ruling. + self.accepted: list[str] = [] + self.rulings: list[JsonDict] = [] + # When the first unruled commit arrived, if a ruling is due. + self.pending_since: int | None = None + + +class Arbiter: + def __init__(self, hs: "HomeServer"): + self.clock = hs.get_clock() + self.server_name = hs.hostname + self.signing_key = hs.signing_key + self.rooms: LruCache[str, ArbiterRoom] = LruCache( + max_size=MAX_ROOMS, + clock=self.clock, + server_name=self.server_name, + cache_name="fermion_arbiter_rooms", + ) + + def room(self, room_id: str) -> ArbiterRoom: + room = self.rooms.get(room_id) + if room is None: + room = ArbiterRoom() + self.rooms[room_id] = room + return room + + def submit(self, room_id: str, body: JsonDict, origin: str) -> JsonDict: + """Record a commit reported by `origin`; whether it is canonical, + and the ruling if one is minted now.""" + assert_params_in_dict(body, ["base", "epoch", "event_id"]) + base, epoch, event_id = body["base"], body["epoch"], body["event_id"] + if not all(isinstance(v, str) for v in (base, epoch, event_id)): + raise SynapseError(400, "'base', 'epoch' and 'event_id' must be strings") + interval = body.get("epoch_interval_ms", 0) + if not isinstance(interval, int) or interval < 0: + raise SynapseError( + 400, "'epoch_interval_ms' must be a non-negative integer" + ) + room = self.room(room_id) + if room.head is None: + room.head = base + accepted = base == room.head + if accepted: + room.head = epoch + room.accepted.append(event_id) + logger.info( + "fermion arbiter: %s commit %s on %s in %s from %s", + "accepted" if accepted else "rejected", + event_id, + base, + room_id, + origin, + ) + if interval == 0: + if accepted or not room.rulings: + self.rule(room_id, room) + return {"accepted": accepted, "ruling": room.rulings[-1]} + if accepted and room.pending_since is None: + room.pending_since = self.clock.time_msec() + self.clock.call_later( + Duration(milliseconds=interval), self.rule, room_id, room + ) + return {"accepted": accepted, "ruling": None} + + def rule(self, room_id: str, room: ArbiterRoom) -> None: + """Mint a ruling: the canonical extremity and the commits accepted + since the last one, signed by this server.""" + ruling = { + "room_id": room_id, + "index": len(room.rulings), + "canonical": room.head, + "commits": room.accepted, + "ts": self.clock.time_msec(), + } + room.rulings.append(sign_json(ruling, self.server_name, self.signing_key)) + del room.rulings[:-RULINGS_KEPT] + room.accepted = [] + room.pending_since = None + + def rulings(self, room_id: str, since: int) -> list[JsonDict]: + room = self.rooms.get(room_id) + if room is None: + return [] + return [r for r in room.rulings if r["index"] >= since] + + +class ArbiterCommitsServlet(servlet.RestServlet): + """`POST …/rooms/{roomId}/commits` with `{arbiter, base, epoch, event_id, + epoch_interval_ms}`: report a commit to the room's arbiter server.""" + + PATTERNS = client_patterns(PREFIX + "/commits$", releases=(), unstable=True) + CATEGORY = "Encryption requests" + + def __init__(self, hs: "HomeServer", arbiter: Arbiter): + super().__init__() + self.hs = hs + self.auth = hs.get_auth() + self.arbiter = arbiter + self.federation = hs.get_federation_http_client() + + async def on_POST( + self, request: SynapseRequest, room_id: str + ) -> tuple[int, JsonDict]: + requester = await self.auth.get_user_by_req(request) + body = parse_json_object_from_request(request) + assert_params_in_dict(body, ["arbiter"]) + destination = body.pop("arbiter") + if not isinstance(destination, str): + raise SynapseError(400, "'arbiter' must be a server name") + body["sender"] = requester.user.to_string() + if destination == self.hs.hostname: + return 200, self.arbiter.submit(room_id, body, self.hs.hostname) + return 200, await self.federation.put_json( + destination, FEDERATION_PATH + room_id + "/commits", data=body + ) + + +class ArbiterRulingsServlet(servlet.RestServlet): + """`GET …/rooms/{roomId}/rulings?arbiter=&from=`: the room's rulings + since index `from`.""" + + PATTERNS = client_patterns(PREFIX + "/rulings$", releases=(), unstable=True) + CATEGORY = "Encryption requests" + + def __init__(self, hs: "HomeServer", arbiter: Arbiter): + super().__init__() + self.hs = hs + self.auth = hs.get_auth() + self.arbiter = arbiter + self.federation = hs.get_federation_http_client() + + async def on_GET( + self, request: SynapseRequest, room_id: str + ) -> tuple[int, JsonDict]: + await self.auth.get_user_by_req(request) + destination = parse_string(request, "arbiter", required=True) + since = parse_integer(request, "from", default=0) + if destination == self.hs.hostname: + return 200, {"rulings": self.arbiter.rulings(room_id, since)} + return 200, await self.federation.get_json( + destination, + FEDERATION_PATH + room_id + "/rulings", + args={"from": str(since)}, + ) + + +def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: + if not hs.config.experimental.fermion_arbiter_enabled: + return + arbiter = hs.get_fermion_arbiter() + ArbiterCommitsServlet(hs, arbiter).register(http_server) + ArbiterRulingsServlet(hs, arbiter).register(http_server) diff --git a/synapse/server.py b/synapse/server.py index 266440534c..6fdbc6e196 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -198,6 +198,7 @@ if TYPE_CHECKING: from synapse.handlers.jwt import JwtHandler from synapse.handlers.oidc import OidcHandler from synapse.handlers.saml import SamlHandler + from synapse.rest.client.fermion import Arbiter as FermionArbiter from synapse.storage._base import SQLBaseStore @@ -889,6 +890,12 @@ class HomeServer(metaclass=abc.ABCMeta): return DeviceHandler(self) + @cache_in_self + def get_fermion_arbiter(self) -> "FermionArbiter": + from synapse.rest.client.fermion import Arbiter + + return Arbiter(self) + @cache_in_self def get_device_message_handler(self) -> DeviceMessageHandler: return DeviceMessageHandler(self) diff --git a/tests/rest/client/test_fermion.py b/tests/rest/client/test_fermion.py new file mode 100644 index 0000000000..0d4e2797b5 --- /dev/null +++ b/tests/rest/client/test_fermion.py @@ -0,0 +1,171 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 New Vector, 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: +# . +# + +from unittest.mock import AsyncMock + +from signedjson.key import get_verify_key +from signedjson.sign import verify_signed_json + +from twisted.test.proto_helpers import MemoryReactor + +from synapse.rest import admin +from synapse.rest.client import fermion, login +from synapse.server import HomeServer +from synapse.types import JsonDict +from synapse.util.clock import Clock + +from tests.unittest import FederatingHomeserverTestCase + +ROOM = "/_matrix/client/unstable/org.matrix.experimental.fermion.v1/rooms/!r:test" +FED_ROOM = ( + "/_matrix/federation/unstable/org.matrix.experimental.fermion.v1/rooms/!r:test" +) + + +class FermionArbiterTestCase(FederatingHomeserverTestCase): + """The arbiter orders the commits members report and signs rulings; + a member's server relays to the room's arbiter server.""" + + servlets = [ + admin.register_servlets, + login.register_servlets, + fermion.register_servlets, + ] + + def default_config(self) -> JsonDict: + config = super().default_config() + config.setdefault("experimental_features", {"fermion_arbiter_enabled": True}) + return config + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + super().prepare(reactor, clock, hs) + self.hs = hs + self.register_user("a", "pass") + self.tok = self.login("a", "pass") + + def submit( + self, + base: str, + epoch: str, + event_id: str, + interval: int = 0, + arbiter: str = "test", + ) -> JsonDict: + channel = self.make_request( + "POST", + ROOM + "/commits", + content={ + "arbiter": arbiter, + "base": base, + "epoch": epoch, + "event_id": event_id, + "epoch_interval_ms": interval, + }, + access_token=self.tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + return channel.json_body + + def rulings(self, since: int = 0) -> list[JsonDict]: + channel = self.make_request( + "GET", ROOM + f"/rulings?arbiter=test&from={since}", access_token=self.tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + return channel.json_body["rulings"] + + def test_sequencer_accepts_the_first_commit_on_the_head(self) -> None: + # The first submission seeds the head with its base. + first = self.submit("e0", "e1", "$c1") + self.assertTrue(first["accepted"]) + self.assertEqual(first["ruling"]["canonical"], "e1") + self.assertEqual(first["ruling"]["commits"], ["$c1"]) + self.assertEqual(first["ruling"]["index"], 0) + verify_signed_json( + first["ruling"], self.hs.hostname, get_verify_key(self.hs.signing_key) + ) + + # A concurrent commit on e0 loses, and learns the head from the ruling. + loser = self.submit("e0", "e1x", "$c1x") + self.assertFalse(loser["accepted"]) + self.assertEqual(loser["ruling"]["canonical"], "e1") + self.assertEqual(loser["ruling"]["index"], 0) + + # Building on the head wins again. + second = self.submit("e1", "e2", "$c2") + self.assertTrue(second["accepted"]) + self.assertEqual(second["ruling"]["canonical"], "e2") + self.assertEqual(second["ruling"]["commits"], ["$c2"]) + self.assertEqual([r["index"] for r in self.rulings()], [0, 1]) + self.assertEqual([r["index"] for r in self.rulings(since=1)], [1]) + + def test_interval_batches_commits_into_one_ruling(self) -> None: + self.assertEqual(self.submit("e0", "e1", "$c1", interval=5000)["ruling"], None) + self.assertEqual(self.submit("e1", "e2", "$c2", interval=5000)["ruling"], None) + fork = self.submit("e1", "e2x", "$c2x", interval=5000) + self.assertFalse(fork["accepted"]) + self.assertEqual(self.rulings(), []) + self.reactor.advance(5) + (ruling,) = self.rulings() + self.assertEqual(ruling["canonical"], "e2") + self.assertEqual(ruling["commits"], ["$c1", "$c2"]) + # The loser rebases: a resolution on the head is canonical. + self.assertTrue(self.submit("e2", "e3", "$c3", interval=5000)["accepted"]) + self.reactor.advance(5) + self.assertEqual(self.rulings(since=1)[0]["commits"], ["$c3"]) + + def test_over_federation(self) -> None: + # Another server's member reports to us, the arbiter. + channel = self.make_signed_federation_request( + "PUT", + FED_ROOM + "/commits", + content={ + "base": "e0", + "epoch": "e1", + "event_id": "$c1", + "sender": "@b:other", + }, + ) + self.assertEqual(channel.code, 200, channel.json_body) + self.assertTrue(channel.json_body["accepted"]) + channel = self.make_signed_federation_request( + "GET", FED_ROOM + "/rulings?from=0" + ) + self.assertEqual(channel.code, 200, channel.json_body) + self.assertEqual(channel.json_body["rulings"][0]["commits"], ["$c1"]) + + # Our member reports to another server's arbiter: relayed, with the + # sender filled in. + client = self.hs.get_federation_http_client() + client.put_json = AsyncMock(return_value={"accepted": True, "ruling": None}) # type: ignore[method-assign] + self.assertTrue(self.submit("x0", "x1", "$d1", arbiter="other")["accepted"]) + destination, path = client.put_json.call_args.args + self.assertEqual(destination, "other") + self.assertEqual(path, FED_ROOM + "/commits") + body = client.put_json.call_args.kwargs["data"] + self.assertEqual(body["sender"], "@a:test") + self.assertNotIn("arbiter", body) + + def test_bad_submissions(self) -> None: + for content in ( + {"base": "e0"}, + {"arbiter": "test", "base": "e0"}, + {"arbiter": "test", "base": "e0", "epoch": "e1", "event_id": 1}, + ): + channel = self.make_request( + "POST", ROOM + "/commits", content=content, access_token=self.tok + ) + self.assertEqual(channel.code, 400, content) + channel = self.make_request("POST", ROOM + "/commits", content={}) + self.assertEqual(channel.code, 401) + self.assertEqual(self.rulings(), [])