Cortex-Inara/cortex/notification.py

"""
Outbound notification helpers — send messages to user channels proactively.

Channel config lives in home/{user}/channels.json.
Each channel that supports proactive notifications needs a notification_channel
set to its key name (e.g. "nextcloud", "google_chat") in the user's channels.json:
  {
    "notification_channel": "nextcloud",
    "nextcloud": {
      "url": "https://cloud.example.com",
      "bot_secret": "...",
      "notification_room": "<room-token>",
      ...
    }
  }

If notification_channel is absent, defaults to "nextcloud" if configured.
If notification_room (for NCT) is absent, notifications are silently skipped.
"""
import hashlib
import hmac
import json
import logging
import secrets

import httpx

logger = logging.getLogger(__name__)


async def _send_nct_message(url: str, secret: str, room: str, message: str) -> None:
    """Post a message to a Nextcloud Talk room as the bot."""
    endpoint = f"{url}/ocs/v2.php/apps/spreed/api/v1/bot/{room}/message"
    random_str = secrets.token_hex(32)
    sig = hmac.new(
        secret.encode(),
        (random_str + message).encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    body = json.dumps({"message": message}, ensure_ascii=False).encode("utf-8")

    try:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                endpoint,
                content=body,
                headers={
                    "Content-Type": "application/json",
                    "OCS-APIRequest": "true",
                    "X-Nextcloud-Talk-Bot-Random": random_str,
                    "X-Nextcloud-Talk-Bot-Signature": sig,
                },
                timeout=15,
            )
        if resp.status_code not in (200, 201):
            logger.warning("notify NCT %s → HTTP %d: %s", room, resp.status_code, resp.text[:200])
        else:
            logger.info("notify NCT → %s (%d chars)", room, len(message))
    except Exception as e:
        logger.error("notify NCT error: %s", e)


async def _notify_nct(nct: dict, message: str, username: str) -> None:
    room   = nct.get("notification_room", "").strip()
    url    = nct.get("url", "").rstrip("/")
    secret = nct.get("bot_secret", "")
    if not room:
        logger.debug("notify: NCT notification_room not set for %s — skipping", username)
        return
    if not url or not secret:
        logger.warning("notify: NCT config incomplete for %s (missing url or secret)", username)
        return
    await _send_nct_message(url, secret, room, message)


async def notify(username: str, message: str, channel: str | None = None) -> None:
    """Send a notification to the user's preferred outbound channel.

    Channel resolution order:
      1. `channel` parameter if provided
      2. `notification_channel` key in channels.json
      3. "nextcloud" if configured
      4. Silent no-op

    To configure: set `notification_channel` in home/{user}/channels.json.
    For NCT: also set `notification_room` in the nextcloud section.
    """
    from auth_utils import get_user_channels
    channels = get_user_channels(username)

    target = channel or channels.get("notification_channel", "").strip()
    if not target:
        # Auto-detect: use nextcloud if configured
        if "nextcloud" in channels:
            target = "nextcloud"
        else:
            return

    if target == "nextcloud":
        nct = channels.get("nextcloud")
        if not nct:
            logger.debug("notify: nextcloud not configured for %s", username)
            return
        await _notify_nct(nct, message, username)
    else:
        logger.debug("notify: channel %r not yet supported for outbound (user %s)", target, username)