# 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 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"
#   2. send one or more chunks, each framed as a 4-byte big-endian length + that many
#      data bytes
#   3. send a zero-length chunk (b"\x00\x00\x00\x00") to signal end of stream
#   4. read the reply: "stream: OK\0" (clean) or "stream: <name> FOUND\0" (infected)
#
# Fails open (allows the file through) on any connection/timeout error against clamd,
# 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 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

logger = logging.getLogger(__name__)

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
        self.clamd_host = config["clamd_host"]
        self.clamd_port = config["clamd_port"]
        self.api.register_spam_checker_callbacks(
            check_media_file_for_spam=self.check_media_file_for_spam,
        )

    @staticmethod
    def parse_config(config: dict) -> dict:
        if "clamd_host" not in config or "clamd_port" not in config:
            raise ValueError(
                "clamav_spam_checker config requires 'clamd_host' and 'clamd_port'"
            )
        return config

    async def check_media_file_for_spam(
        self, file_wrapper: Any, file_info: Any
    ) -> Union[Any, Codes, bool]:
        buffer = bytearray()
        await file_wrapper.write_chunks_to(buffer.extend)

        try:
            verdict = await self._scan(bytes(buffer))
        except Exception:
            logger.exception(
                "ClamAV scan failed (clamd at %s:%s unreachable?) - "
                "allowing file through (fail-open)",
                self.clamd_host,
                self.clamd_port,
            )
            return NOT_SPAM

        if verdict is None:
            return NOT_SPAM

        logger.warning("ClamAV rejected an upload: %s", verdict)
        return Codes.FORBIDDEN

    async def _scan(self, data: bytes) -> "str | None":
        """Returns the detected signature name, or None if the file is clean."""
        result: "Deferred[bytes]" = Deferred()
        endpoint = HostnameEndpoint(reactor, self.clamd_host.encode(), self.clamd_port)
        await connectProtocol(endpoint, _ClamdInstreamProtocol(data, result))

        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: <signature name> FOUND"
        if text.endswith("FOUND"):
            return text.removeprefix("stream:").removesuffix("FOUND").strip()
        return None