notifications.py

"""
notifications.py — KevinNotifications
Multi-channel alert system ported from SOMA's KevinNotificationService.cjs and KevinSMSService.cjs.
Channels: Telegram (two-way polling) > Discord (webhook) > Slack (webhook)
SMS via email-to-SMS gateways (13 US carriers from KevinSMSService.cjs).
"""

import asyncio
import json
import time
from typing import Callable, Optional
import aiohttp


# ported from KevinSMSService.cjs CARRIER_GATEWAYS
CARRIER_GATEWAYS = {
    "att":       {"name": "AT&T",          "sms": "txt.att.net",             "mms": "mms.att.net"},
    "verizon":   {"name": "Verizon",       "sms": "vtext.com",               "mms": "vzwpix.com"},
    "tmobile":   {"name": "T-Mobile",      "sms": "tmomail.net",             "mms": "tmomail.net"},
    "sprint":    {"name": "Sprint",        "sms": "messaging.sprintpcs.com", "mms": "pm.sprint.com"},
    "uscellular":{"name": "US Cellular",   "sms": "email.uscc.net",          "mms": "mms.uscc.net"},
    "metropcs":  {"name": "Metro PCS",     "sms": "mymetropcs.com",          "mms": "mymetropcs.com"},
    "boost":     {"name": "Boost Mobile",  "sms": "sms.myboostmobile.com",   "mms": "myboostmobile.com"},
    "cricket":   {"name": "Cricket",       "sms": "sms.cricketwireless.net", "mms": "mms.cricketwireless.net"},
    "virgin":    {"name": "Virgin Mobile", "sms": "vmobl.com",               "mms": "vmpix.com"},
    "googlefi":  {"name": "Google Fi",     "sms": "msg.fi.google.com",       "mms": "msg.fi.google.com"},
    "xfinity":   {"name": "Xfinity",       "sms": "vtext.com",               "mms": "mypixmessages.com"},
    "visible":   {"name": "Visible",       "sms": "vtext.com",               "mms": "vzwpix.com"},
    "mint":      {"name": "Mint Mobile",   "sms": "tmomail.net",             "mms": "tmomail.net"},
}

SEVERITY_EMOJI = {"low": "ℹ️", "medium": "⚠️", "high": "🚨", "critical": "🔥"}


class KevinNotifications:
    def __init__(self, config: dict):
        """
        config keys:
          telegram_token, telegram_chat_id
          discord_webhook
          slack_webhook
          sms_phone, sms_carrier  (optional)
        """
        self._telegram_token = config.get("telegram_token", "")
        self._telegram_chat_id = config.get("telegram_chat_id", "")
        self._discord_webhook = config.get("discord_webhook", "")
        self._slack_webhook = config.get("slack_webhook", "")
        self._sms_phone = config.get("sms_phone", "")
        self._sms_carrier = config.get("sms_carrier", "")

        self._telegram_offset = 0
        self._polling_active = False
        self._session: Optional[aiohttp.ClientSession] = None

        # Rate limiting: max 20/hour per channel
        self._rate: dict = {}

    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=15)
            )
        return self._session

    def _check_rate(self, channel: str, limit: int = 20) -> bool:
        now = time.time()
        entry = self._rate.get(channel, {"count": 0, "reset_at": now + 3600})
        if now > entry["reset_at"]:
            entry = {"count": 0, "reset_at": now + 3600}
        if entry["count"] >= limit:
            return False
        entry["count"] += 1
        self._rate[channel] = entry
        return True

    # ------------------------------------------------------------------ #
    # Public API
    # ------------------------------------------------------------------ #

    async def send_alert(self, title: str, message: str, severity: str = "low"):
        """Send to all configured channels. severity: low|medium|high|critical"""
        emoji = SEVERITY_EMOJI.get(severity, "🛡️")
        full_title = f"{emoji} {title}"
        tasks = []
        if self._telegram_token and self._telegram_chat_id:
            tasks.append(self._send_telegram_message(full_title, message, severity))
        if self._discord_webhook:
            tasks.append(self._send_discord(full_title, message, severity))
        if self._slack_webhook:
            tasks.append(self._send_slack(full_title, message, severity))
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

    async def send_threat_alert(self, subject: str, sender: str, reason: str):
        await self.send_alert(
            title=f"Threat Detected: {subject[:60]}",
            message=f"From: {sender}\nReason: {reason}",
            severity="high"
        )

    async def send_morning_briefing(self, summary: str):
        await self.send_alert(
            title="Kevin Morning Briefing",
            message=summary,
            severity="low"
        )

    # ------------------------------------------------------------------ #
    # Telegram
    # ------------------------------------------------------------------ #

    async def _send_telegram_message(self, title: str, body: str, severity: str):
        if not self._check_rate("telegram"):
            return
        text = f"<b>{title}</b>\n\n{body}\n\n<i>🤖 KEVIN Security</i>"
        url = f"https://api.telegram.org/bot{self._telegram_token}/sendMessage"
        payload = {
            "chat_id": self._telegram_chat_id,
            "text": text,
            "parse_mode": "HTML",
            "disable_web_page_preview": True
        }
        try:
            session = await self._get_session()
            async with session.post(url, json=payload) as resp:
                pass  # fire and forget
        except Exception:
            pass

    async def send_telegram_text(self, text: str):
        """Send raw text to Telegram (for command responses)."""
        if not (self._telegram_token and self._telegram_chat_id):
            return
        url = f"https://api.telegram.org/bot{self._telegram_token}/sendMessage"
        try:
            session = await self._get_session()
            await session.post(url, json={
                "chat_id": self._telegram_chat_id,
                "text": text,
                "parse_mode": "HTML"
            })
        except Exception:
            pass

    async def start_telegram_polling(self, command_handler: Callable[[str], None]):
        """
        Poll /getUpdates every 3 seconds.
        Calls command_handler(text) for each received message.
        Allowed commands: scan, status, threats, summary, pause, resume
        """
        if not (self._telegram_token and self._telegram_chat_id):
            print("[Notifications] Telegram not configured — polling skipped")
            return

        self._polling_active = True
        print("[Notifications] Telegram polling started")
        allowed_cmds = {"scan", "status", "threats", "summary", "pause", "resume"}

        while self._polling_active:
            try:
                url = (
                    f"https://api.telegram.org/bot{self._telegram_token}"
                    f"/getUpdates?offset={self._telegram_offset + 1}&timeout=2"
                )
                session = await self._get_session()
                async with session.get(url) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        if data.get("ok"):
                            for update in data.get("result", []):
                                self._telegram_offset = update["update_id"]
                                msg = update.get("message", {})
                                text = msg.get("text", "").strip().lower().lstrip("/")
                                if text in allowed_cmds:
                                    try:
                                        result = command_handler(text)
                                        if asyncio.iscoroutine(result):
                                            await result
                                    except Exception as e:
                                        print(f"[Notifications] Command handler error: {e}")
            except Exception:
                pass
            await asyncio.sleep(3)

    def stop_polling(self):
        self._polling_active = False

    # ------------------------------------------------------------------ #
    # Discord
    # ------------------------------------------------------------------ #

    async def _send_discord(self, title: str, message: str, severity: str):
        if not self._discord_webhook or not self._check_rate("discord"):
            return
        colors = {"low": 0x3498db, "medium": 0xf39c12, "high": 0xe74c3c, "critical": 0x9b59b6}
        payload = {
            "username": "KEVIN Security",
            "embeds": [{
                "title": title,
                "description": message,
                "color": colors.get(severity, 0x95a5a6),
                "footer": {"text": "KEVIN Security System"}
            }]
        }
        try:
            session = await self._get_session()
            async with session.post(self._discord_webhook, json=payload) as resp:
                pass
        except Exception:
            pass

    # ------------------------------------------------------------------ #
    # Slack
    # ------------------------------------------------------------------ #

    async def _send_slack(self, title: str, message: str, severity: str):
        if not self._slack_webhook or not self._check_rate("slack"):
            return
        payload = {
            "blocks": [
                {"type": "header", "text": {"type": "plain_text", "text": title, "emoji": True}},
                {"type": "section", "text": {"type": "mrkdwn", "text": message}},
                {"type": "context", "elements": [{"type": "mrkdwn", "text": ":robot_face: KEVIN Security"}]}
            ]
        }
        try:
            session = await self._get_session()
            async with session.post(self._slack_webhook, json=payload) as resp:
                pass
        except Exception:
            pass

    # ------------------------------------------------------------------ #
    # SMS (email-to-SMS gateway)
    # ------------------------------------------------------------------ #

    def get_sms_address(self) -> Optional[str]:
        """Returns the email address for SMS gateway, or None if not configured."""
        if not self._sms_phone or not self._sms_carrier:
            return None
        gateway = CARRIER_GATEWAYS.get(self._sms_carrier, {}).get("sms")
        if not gateway:
            return None
        # Normalize phone: 10 digits only
        digits = "".join(c for c in self._sms_phone if c.isdigit())
        if len(digits) == 11 and digits.startswith("1"):
            digits = digits[1:]
        if len(digits) != 10:
            return None
        return f"{digits}@{gateway}"

    # ------------------------------------------------------------------ #
    # Status
    # ------------------------------------------------------------------ #

    def get_status(self) -> dict:
        return {
            "telegram": bool(self._telegram_token and self._telegram_chat_id),
            "discord": bool(self._discord_webhook),
            "slack": bool(self._slack_webhook),
            "sms": bool(self.get_sms_address()),
            "sms_address": self.get_sms_address(),
            "polling_active": self._polling_active
        }

    async def close(self):
        self._polling_active = False
        if self._session and not self._session.closed:
            await self._session.close()