From dc856186d1d84209f042501d11f6a22a52da1bf3 Mon Sep 17 00:00:00 2001 From: Thore Cimbal Date: Wed, 29 Jul 2026 15:42:46 +0200 Subject: [PATCH] fix: use Twisted networking instead of asyncio in ClamAV module Synapse runs on Twisted's reactor, not asyncio's event loop - the original asyncio.open_connection/wait_for calls failed immediately with "RuntimeError: no running event loop", silently fail-opening every scan (confirmed live: EICAR test file passed through unscanned). Rewritten using twisted.internet.endpoints.HostnameEndpoint/ connectProtocol and a custom Protocol for the INSTREAM conversation. --- apps/production/clamav_spam_checker.py | 67 ++++++++++++++++++-------- 1 file changed, 48 insertions(+), 19 deletions(-) diff --git a/apps/production/clamav_spam_checker.py b/apps/production/clamav_spam_checker.py index 0563971..00aa554 100644 --- a/apps/production/clamav_spam_checker.py +++ b/apps/production/clamav_spam_checker.py @@ -1,9 +1,13 @@ # Synapse spam-checker module (Issue #19): scans locally-stored and remote/federated media # through ClamAV's clamd daemon via its native INSTREAM protocol, before Synapse serves it. # -# Talks to clamd using only the standard library (asyncio/socket) - no third-party -# dependency, since Synapse's container runs with a read-only root filesystem and this -# module is mounted in via a ConfigMap, not baked into a custom image. +# Talks to clamd using Twisted's networking primitives - NOT asyncio's open_connection/ +# wait_for. Synapse runs on Twisted's reactor, which does not have a running asyncio event +# loop underneath it, so raw asyncio socket calls fail immediately with +# "RuntimeError: no running event loop" (confirmed live, 2026-07-29 - see git history for +# the asyncio-based version that failed this way). Twisted Deferreds are natively awaitable +# from an `async def` when Synapse wraps the callback via Deferred.fromCoroutine(), so this +# stays plain async/await from the caller's perspective. # # clamd INSTREAM protocol (docs.clamav.net/manual/Usage/ClamdProtocol.html): # 1. send b"zINSTREAM\0" @@ -16,10 +20,14 @@ # so a scanner outage can't take down media uploads for the whole homeserver - logged # loudly so an outage is still visible in the logs. -import asyncio import logging from typing import Any, Union +from twisted.internet import reactor +from twisted.internet.defer import Deferred, TimeoutError as TwistedTimeoutError +from twisted.internet.endpoints import HostnameEndpoint, connectProtocol +from twisted.internet.protocol import Protocol + from synapse.module_api import ModuleApi, NOT_SPAM from synapse.module_api.errors import Codes @@ -29,6 +37,31 @@ CHUNK_SIZE = 2**14 # matches ReadableFileWrapper.CHUNK_SIZE CLAMD_TIMEOUT_SECONDS = 30 +class _ClamdInstreamProtocol(Protocol): + """Speaks clamd's INSTREAM protocol for a single scan, then closes.""" + + def __init__(self, data: bytes, result: "Deferred[bytes]"): + self._data = data + self._result = result + self._buffer = bytearray() + + def connectionMade(self) -> None: + self.transport.write(b"zINSTREAM\0") + for offset in range(0, len(self._data), CHUNK_SIZE): + chunk = self._data[offset : offset + CHUNK_SIZE] + self.transport.write(len(chunk).to_bytes(4, "big") + chunk) + self.transport.write((0).to_bytes(4, "big")) + + def dataReceived(self, data: bytes) -> None: + self._buffer.extend(data) + if self._buffer.endswith(b"\0") or self._buffer.endswith(b"\n"): + self.transport.loseConnection() + + def connectionLost(self, reason: Any = None) -> None: + if not self._result.called: + self._result.callback(bytes(self._buffer)) + + class ClamAVSpamChecker: def __init__(self, config: dict, api: ModuleApi): self.api = api @@ -53,9 +86,7 @@ class ClamAVSpamChecker: await file_wrapper.write_chunks_to(buffer.extend) try: - verdict = await asyncio.wait_for( - self._scan(bytes(buffer)), timeout=CLAMD_TIMEOUT_SECONDS - ) + verdict = await self._scan(bytes(buffer)) except Exception: logger.exception( "ClamAV scan failed (clamd at %s:%s unreachable?) - " @@ -73,19 +104,17 @@ class ClamAVSpamChecker: async def _scan(self, data: bytes) -> "str | None": """Returns the detected signature name, or None if the file is clean.""" - reader, writer = await asyncio.open_connection(self.clamd_host, self.clamd_port) - try: - writer.write(b"zINSTREAM\0") - for offset in range(0, len(data), CHUNK_SIZE): - chunk = data[offset : offset + CHUNK_SIZE] - writer.write(len(chunk).to_bytes(4, "big") + chunk) - writer.write((0).to_bytes(4, "big")) - await writer.drain() + result: "Deferred[bytes]" = Deferred() + endpoint = HostnameEndpoint(reactor, self.clamd_host.encode(), self.clamd_port) + await connectProtocol(endpoint, _ClamdInstreamProtocol(data, result)) - response = await reader.read() - finally: - writer.close() - await writer.wait_closed() + result.addTimeout(CLAMD_TIMEOUT_SECONDS, reactor) + try: + response = await result + except TwistedTimeoutError: + raise TimeoutError( + f"clamd at {self.clamd_host}:{self.clamd_port} did not respond in time" + ) text = response.decode("utf-8", errors="replace").strip("\x00 \n") # "stream: OK" or "stream: FOUND"