fix: use Twisted networking instead of asyncio in ClamAV module
Auto-Deploy on Push / verify-and-notify (push) Canceled after 0s

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.
This commit is contained in:
Thore Cimbal
2026-07-29 15:42:46 +02:00
parent db77785b6b
commit dc856186d1
+48 -19
View File
@@ -1,9 +1,13 @@
# Synapse spam-checker module (Issue #19): scans locally-stored and remote/federated media # 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. # 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 # Talks to clamd using Twisted's networking primitives - NOT asyncio's open_connection/
# dependency, since Synapse's container runs with a read-only root filesystem and this # wait_for. Synapse runs on Twisted's reactor, which does not have a running asyncio event
# module is mounted in via a ConfigMap, not baked into a custom image. # 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): # clamd INSTREAM protocol (docs.clamav.net/manual/Usage/ClamdProtocol.html):
# 1. send b"zINSTREAM\0" # 1. send b"zINSTREAM\0"
@@ -16,10 +20,14 @@
# so a scanner outage can't take down media uploads for the whole homeserver - logged # 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. # loudly so an outage is still visible in the logs.
import asyncio
import logging import logging
from typing import Any, Union 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 import ModuleApi, NOT_SPAM
from synapse.module_api.errors import Codes from synapse.module_api.errors import Codes
@@ -29,6 +37,31 @@ CHUNK_SIZE = 2**14 # matches ReadableFileWrapper.CHUNK_SIZE
CLAMD_TIMEOUT_SECONDS = 30 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: class ClamAVSpamChecker:
def __init__(self, config: dict, api: ModuleApi): def __init__(self, config: dict, api: ModuleApi):
self.api = api self.api = api
@@ -53,9 +86,7 @@ class ClamAVSpamChecker:
await file_wrapper.write_chunks_to(buffer.extend) await file_wrapper.write_chunks_to(buffer.extend)
try: try:
verdict = await asyncio.wait_for( verdict = await self._scan(bytes(buffer))
self._scan(bytes(buffer)), timeout=CLAMD_TIMEOUT_SECONDS
)
except Exception: except Exception:
logger.exception( logger.exception(
"ClamAV scan failed (clamd at %s:%s unreachable?) - " "ClamAV scan failed (clamd at %s:%s unreachable?) - "
@@ -73,19 +104,17 @@ class ClamAVSpamChecker:
async def _scan(self, data: bytes) -> "str | None": async def _scan(self, data: bytes) -> "str | None":
"""Returns the detected signature name, or None if the file is clean.""" """Returns the detected signature name, or None if the file is clean."""
reader, writer = await asyncio.open_connection(self.clamd_host, self.clamd_port) result: "Deferred[bytes]" = Deferred()
try: endpoint = HostnameEndpoint(reactor, self.clamd_host.encode(), self.clamd_port)
writer.write(b"zINSTREAM\0") await connectProtocol(endpoint, _ClamdInstreamProtocol(data, result))
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()
response = await reader.read() result.addTimeout(CLAMD_TIMEOUT_SECONDS, reactor)
finally: try:
writer.close() response = await result
await writer.wait_closed() 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") text = response.decode("utf-8", errors="replace").strip("\x00 \n")
# "stream: OK" or "stream: <signature name> FOUND" # "stream: OK" or "stream: <signature name> FOUND"