Add support for receiving EDUs directly (MSC2409)

mautrix/appservice/appservice.py

@@ -11,6 +11,7 @@
 import logging
 
 from mautrix.types import JSON, UserID, RoomAlias
+from mautrix.util.logging import TraceLogger
 
 from .api import AppServiceAPI, IntentAPI
 from .state_store import ASStateStore, FileASStateStore
@@ -47,7 +48,7 @@ class AppService(AppServiceServerMixin):
     live: bool
 
     loop: asyncio.AbstractEventLoop
-    log: logging.Logger
+    log: TraceLogger
     app: web.Application
     runner: web.AppRunner
 
@@ -57,8 +58,9 @@ def __init__(self, server: str, domain: str, as_token: str, hs_token: str, bot_l
                  tls_cert: Optional[str] = None, tls_key: Optional[str] = None,
                  query_user: QueryFunc = None, query_alias: QueryFunc = None,
                  real_user_content_key: Optional[str] = "net.maunium.appservice.puppet",
-                 state_store: ASStateStore = None, aiohttp_params: Dict = None) -> None:
-        super().__init__()
+                 state_store: ASStateStore = None, aiohttp_params: Dict = None,
+                 ephemeral_events: bool = False) -> None:
+        super().__init__(ephemeral_events=ephemeral_events)
         self.server = server
         self.domain = domain
         self.id = id

mautrix/appservice/as_handler.py

@@ -10,7 +10,7 @@
 import asyncio
 import logging
 
-from mautrix.types import JSON, UserID, RoomAlias, Event, SerializerError
+from mautrix.types import JSON, UserID, RoomAlias, Event, EphemeralEvent, SerializerError
 
 QueryFunc = Callable[[web.Request], Awaitable[Optional[web.Response]]]
 HandlerFunc = Callable[[Event], Awaitable]
@@ -21,16 +21,18 @@ class AppServiceServerMixin:
     log: logging.Logger
 
     hs_token: str
+    ephemeral_events: bool
 
     query_user: Callable[[UserID], JSON]
     query_alias: Callable[[RoomAlias], JSON]
 
     transactions: Set[str]
     event_handlers: List[HandlerFunc]
 
-    def __init__(self) -> None:
+    def __init__(self, ephemeral_events: bool = False) -> None:
         self.transactions = set()
         self.event_handlers = []
+        self.ephemeral_events = ephemeral_events
 
         async def default_query_handler(_):
             return None
@@ -64,45 +66,45 @@ def _check_token(self, request: web.Request) -> bool:
 
     async def _http_query_user(self, request: web.Request) -> web.Response:
         if not self._check_token(request):
-            return web.Response(status=401)
+            return web.json_response({"error": "Invalid auth token"}, status=401)
 
         try:
             user_id = request.match_info["user_id"]
         except KeyError:
-            return web.Response(status=400)
+            return web.json_response({"error": "Missing user_id parameter"}, status=400)
 
         try:
             response = await self.query_user(user_id)
         except Exception:
             self.log.exception("Exception in user query handler")
-            return web.Response(status=500)
+            return web.json_response({"error": "Internal appservice error"}, status=500)
 
         if not response:
-            return web.Response(status=404)
+            return web.json_response({}, status=404)
         return web.json_response(response)
 
     async def _http_query_alias(self, request: web.Request) -> web.Response:
         if not self._check_token(request):
-            return web.Response(status=401)
+            return web.json_response({"error": "Invalid auth token"}, status=401)
 
         try:
             alias = request.match_info["alias"]
         except KeyError:
-            return web.Response(status=400)
+            return web.json_response({"error": "Missing alias parameter"}, status=400)
 
         try:
             response = await self.query_alias(alias)
         except Exception:
             self.log.exception("Exception in alias query handler")
-            return web.Response(status=500)
+            return web.json_response({"error": "Internal appservice error"}, status=500)
 
         if not response:
-            return web.Response(status=404)
+            return web.json_response({}, status=404)
         return web.json_response(response)
 
     async def _http_handle_transaction(self, request: web.Request) -> web.Response:
         if not self._check_token(request):
-            return web.Response(status=401)
+            return web.json_response({"error": "Invalid auth token"}, status=401)
 
         transaction_id = request.match_info["transaction_id"]
         if transaction_id in self.transactions:
@@ -111,15 +113,26 @@ async def _http_handle_transaction(self, request: web.Request) -> web.Response:
         try:
             json = await request.json()
         except JSONDecodeError:
-            return web.Response(status=400)
+            return web.json_response({"error": "Body is not JSON"}, status=400)
 
         try:
             events = json["events"]
         except KeyError:
-            return web.Response(status=400)
+            return web.json_response({"error": "Missing events object in body"}, status=400)
+
+        if self.ephemeral_events:
+            try:
+                ephemeral = json["ephemeral"]
+            except KeyError:
+                try:
+                    ephemeral = json["de.sorunome.msc2409.ephemeral"]
+                except KeyError:
+                    ephemeral = None
+        else:
+            ephemeral = None
 
         try:
-            await self.handle_transaction(transaction_id, events)
+            await self.handle_transaction(transaction_id, events=events, ephemeral=ephemeral)
         except Exception:
             self.log.exception("Exception in transaction handler")
 
@@ -137,7 +150,15 @@ def _fix_prev_content(raw_event: JSON) -> None:
             except KeyError:
                 pass
 
-    async def handle_transaction(self, txn_id: str, events: List[JSON]) -> None:
+    async def handle_transaction(self, txn_id: str, events: List[JSON],
+                                 ephemeral: Optional[List[JSON]] = None) -> None:
+        for raw_edu in ephemeral:
+            try:
+                edu = EphemeralEvent.deserialize(raw_edu)
+            except SerializerError:
+                self.log.exception("Failed to deserialize ephemeral event %s", raw_edu)
+            else:
+                self.handle_matrix_event(edu)
         for raw_event in events:
             try:
                 self._fix_prev_content(raw_event)

mautrix/bridge/commands/meta.py

@@ -11,7 +11,7 @@
                       SECTION_GENERAL)
 
 
-@command_handler(help_section=SECTION_GENERAL,
+@command_handler(needs_auth=False, help_section=SECTION_GENERAL,
                  help_text="Cancel an ongoing action.")
 async def cancel(evt: CommandEvent) -> EventID:
     if evt.sender.command_status:
@@ -22,7 +22,7 @@ async def cancel(evt: CommandEvent) -> EventID:
         return await evt.reply("No ongoing command.")
 
 
-@command_handler(help_section=SECTION_GENERAL,
+@command_handler(needs_auth=False, help_section=SECTION_GENERAL,
                  help_text="Get the bridge version.")
 async def version(evt: CommandEvent) -> None:
     if not evt.processor.bridge:
@@ -32,7 +32,7 @@ async def version(evt: CommandEvent) -> None:
                         f"{evt.processor.bridge.markdown_version or evt.processor.bridge.version}")
 
 
-@command_handler()
+@command_handler(needs_auth=False)
 async def unknown_command(evt: CommandEvent) -> EventID:
     return await evt.reply("Unknown command. Try `$cmdprefix+sp help` for help.")
 
@@ -63,8 +63,7 @@ def _get_management_status(evt: CommandEvent) -> str:
     return "**This is not a management room**: you must prefix commands with `$cmdprefix`."
 
 
-@command_handler(name="help",
-                 help_section=SECTION_GENERAL,
+@command_handler(name="help", needs_auth=False, help_section=SECTION_GENERAL,
                  help_text="Show this help message.")
 async def help_cmd(evt: CommandEvent) -> EventID:
     return await evt.reply(_get_management_status(evt) + "\n" + await _get_help_text(evt))

mautrix/bridge/config.py

@@ -74,6 +74,8 @@ def do_update(self, helper: ConfigUpdateHelper) -> None:
         copy("appservice.as_token")
         copy("appservice.hs_token")
 
+        copy("appservice.ephemeral_events")
+
         copy("logging")
 
     @property
@@ -123,3 +125,7 @@ def generate_registration(self) -> None:
             "sender_localpart": self._new_token(),
             "rate_limited": False
         }
+
+        if self["appservice.ephemeral_events"]:
+            self._registration["de.sorunome.msc2409.push_ephemeral"] = True
+            self._registration["push_ephemeral"] = True

mautrix/bridge/matrix.py

@@ -426,6 +426,9 @@ async def int_handle_event(self, evt: Event) -> None:
         else:
             if evt.type.is_state and isinstance(evt, StateEvent):
                 await self.handle_state_event(evt)
+            elif evt.type.is_ephemeral and isinstance(evt, (PresenceEvent, TypingEvent,
+                                                            ReceiptEvent)):
+                await self.handle_ephemeral_event(evt)
             else:
                 await self.handle_event(evt)