srs-api/hermes-plugin/srs-gate/__init__.py

from __future__ import annotations

import asyncio
import json
import logging
import os
import time
import urllib.error
import urllib.request
from pathlib import Path

log = logging.getLogger("srs_gate")

SKIP = {"action": "skip", "reason": "srs-gate"}
CONFIG_NAME = "srs-gate.env"
TRIES = 3  
TRY_TIMEOUT_S = 3.0
BACKOFF_S = 0.3
OVERALL_TIMEOUT_S = 9.0

_config_warned = False


class ConfigError(Exception):
    """srs-gate.env is missing or incomplete. Messages carry fixed wording only, never values."""


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Never follow a redirect: it could carry the x-srs-secret header (and the text) to another host."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None 

_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect)


# ---------------------------------------------------------------- config


def _config_path() -> Path:
    home = os.environ.get("HERMES_HOME")
    base = Path(home) if home else Path.home() / ".hermes"
    return base / CONFIG_NAME


def _unquote(value: str) -> str:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
        return value[1:-1]
    return value


def _load_config() -> dict:
    try:
        raw = _config_path().read_text(encoding="utf-8")
    except FileNotFoundError:
        raise ConfigError("config file not found") from None
    except (OSError, UnicodeError):
        raise ConfigError("config file unreadable") from None
    cfg = {}
    for line in raw.splitlines():
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            key, _, value = line.partition("=")
            cfg[key.strip()] = _unquote(value.strip())
    url = cfg.get("SRS_URL", "")
    secret = cfg.get("SRS_SECRET", "")
    if not secret:
        raise ConfigError("config has no SRS_SECRET")
    if not url.lower().startswith(("http://", "https://")):
        raise ConfigError("config SRS_URL missing or not http(s)")
    return {"SRS_URL": url, "SRS_SECRET": secret}


# ---------------------------------------------------------------- event filter


def _payload(event):
    """The /ingest body for a forwardable direct text message, else None."""
    if event is None:
        return None
    src = getattr(event, "source", None)
    if src is None or getattr(event, "internal", False) or getattr(src, "is_bot", False):
        return None
    if getattr(src, "chat_type", None) != "dm": 
        return None
    if getattr(event, "edit_revision", None) not in (None, ""):
        return None
    mtype = getattr(event, "message_type", None)
    if mtype is None or str(getattr(mtype, "value", mtype)) != "text":
        return None
    message_id = getattr(event, "message_id", None)
    sender = getattr(src, "user_id", None) 
    chat_id = getattr(src, "chat_id", None)
    text = getattr(event, "text", None)
    if not (message_id and sender and chat_id and isinstance(text, str) and text.strip()):
        return None
    if str(chat_id).endswith("@g.us"):
        return None
    return {
        "kind": "text",
        "message_id": str(message_id),
        "sender_jid": str(sender),
        "chat_id": str(chat_id),
        "text": text,
    }


# ---------------------------------------------------------------- HTTP (runs in a worker thread)


def _error_name(exc: BaseException) -> str:
    reason = getattr(exc, "reason", None) if isinstance(exc, urllib.error.URLError) else None
    return type(reason).__name__ if isinstance(reason, BaseException) else type(exc).__name__


def _post(cfg: dict, payload: dict):
    """Blocking POST with retries. Returns (http_status or None, error type name or None). Never raises.

    Nothing may cross the thread boundary: on Python 3.9-3.11 asyncio.wait_for runs this in its own task, and a
    BaseException such as SystemExit would escape through the event loop instead of the callback's except.
    """
    status = None
    error = None
    try:
        req = urllib.request.Request(
            cfg["SRS_URL"].rstrip("/") + "/ingest",
            data=json.dumps(payload).encode("utf-8"),
            method="POST",
            headers={"content-type": "application/json", "x-srs-secret": cfg["SRS_SECRET"]},
        )
        for attempt in range(TRIES):
            try:
                with _OPENER.open(req, timeout=TRY_TIMEOUT_S) as resp:
                    return resp.status, None
            except urllib.error.HTTPError as err:
                status, error = err.code, None
                err.close()
                if err.code < 500:
                    return status, None  
            except Exception as exc: 
                status, error = None, _error_name(exc)
            if attempt + 1 < TRIES:
                time.sleep(BACKOFF_S)
        return status, error
    except BaseException as exc:
        return status, type(exc).__name__


# ---------------------------------------------------------------- the hook


async def on_inbound(event=None, gateway=None, session_store=None, **kwargs):
    global _config_warned
    message_id = "-"
    try:
        payload = _payload(event)
        if payload is not None:
            message_id = payload["message_id"]
            try:
                cfg = _load_config()
            except ConfigError as exc:
                cfg = None
                if not _config_warned:  
                    _config_warned = True
                    log.warning("srs-gate: %s; nothing is forwarded", exc)
            if cfg is not None:
                _config_warned = False
                status, error = await asyncio.wait_for(
                    asyncio.to_thread(_post, cfg, payload), OVERALL_TIMEOUT_S
                )
                if status is not None and 200 <= status < 300:
                    log.info("srs-gate: id=%s status=%s", message_id, status)
                else:
                    log.warning("srs-gate: id=%s status=%s error=%s", message_id, status, error)
    except asyncio.CancelledError:
        raise
    except (Exception, SystemExit) as exc:  
        log.warning("srs-gate: id=%s not forwarded (%s)", message_id, type(exc).__name__)
    return dict(SKIP)


def register(ctx):
    ctx.register_hook("pre_gateway_dispatch", on_inbound)