spam_guard.py

# Copyright (C) 2026 Dariush Lashani
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""سیستم آنتی‌اسپم، rate limiting لغزنده (sliding window) با Redis."""

import os
import time
import logging
from enum import Enum, auto

import redis.asyncio as redis
import metrics

logger = logging.getLogger(__name__)

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
r = redis.from_url(REDIS_URL, decode_responses=True)

# --- تنظیمات ---
MSG_LIMIT  = int(os.environ.get("SPAM_MSG_LIMIT",  "12"))
MSG_WINDOW = int(os.environ.get("SPAM_MSG_WINDOW", "5"))

CMD_LIMIT  = int(os.environ.get("SPAM_CMD_LIMIT",  "8"))
CMD_WINDOW = int(os.environ.get("SPAM_CMD_WINDOW", "10"))

FLOOD_LIMIT  = int(os.environ.get("SPAM_FLOOD_LIMIT",  "30"))
FLOOD_WINDOW = int(os.environ.get("SPAM_FLOOD_WINDOW", "30"))

BLOCK_DURATION = int(os.environ.get("SPAM_BLOCK_DURATION", "60"))

_KEY_BLOCK = "bluechat:spam_block:{user_id}"
_KEY_RATE  = "bluechat:spam_rate:{kind}:{user_id}"
_KEY_FLOOD = "bluechat:spam_flood:{user_id}"


class SpamResult(Enum):
    ALLOWED        = auto()   # مجاز
    JUST_BLOCKED   = auto()   # همین الان بلاک شد، یه هشدار بفرست
    ALREADY_BLOCKED = auto()  # از قبل بلاک بود، سکوت کن (silent drop)


async def _sliding_window(key: str, limit: int, window: int) -> bool:
    """sliding window با Redis. هر بار قدیمی‌ترها رو پاک می‌کنیم که رم
    بیهوده مصرف نشه. True یعنی مجاز، False یعنی از سقف رد شده."""
    now = time.time()
    cutoff = now - window
    async with r.pipeline(transaction=True) as pipe:
        pipe.zremrangebyscore(key, "-inf", cutoff)   # قدیمی‌ها رو پاک کن
        pipe.zadd(key, {str(now): now})               # timestamp جدید
        pipe.zcard(key)                               # شمارش پنجره
        pipe.expire(key, window + 1)                  # TTL خودکار
        _, _, count, _ = await pipe.execute()
    return count <= limit


async def is_blocked(user_id: int) -> bool:
    return bool(await r.exists(_KEY_BLOCK.format(user_id=user_id)))


async def _block_user(user_id: int) -> None:
    await r.setex(_KEY_BLOCK.format(user_id=user_id), BLOCK_DURATION, "1")
    logger.warning("spam_guard: user %s blocked for %ds", user_id, BLOCK_DURATION)


async def check_message(user_id: int) -> SpamResult:
    """بررسی پیام متنی/مدیا. سه حالت داره: ALLOWED (ادامه بده)،
    JUST_BLOCKED (یه هشدار بفرست بعد drop کن)، ALREADY_BLOCKED (بی‌صدا
    drop کن)."""
    if await is_blocked(user_id):
        return SpamResult.ALREADY_BLOCKED

    flood_key = _KEY_FLOOD.format(user_id=user_id)
    msg_key   = _KEY_RATE.format(kind="msg", user_id=user_id)

    flood_ok = await _sliding_window(flood_key, FLOOD_LIMIT, FLOOD_WINDOW)
    msg_ok   = await _sliding_window(msg_key,   MSG_LIMIT,   MSG_WINDOW)

    if not flood_ok or not msg_ok:
        await _block_user(user_id)
        metrics.spam_blocks.labels(kind="message").inc()
        return SpamResult.JUST_BLOCKED

    return SpamResult.ALLOWED


async def check_command(user_id: int) -> SpamResult:
    """بررسی دستور یا callback. همون سه حالت بالا."""
    if await is_blocked(user_id):
        return SpamResult.ALREADY_BLOCKED

    cmd_key = _KEY_RATE.format(kind="cmd", user_id=user_id)
    ok = await _sliding_window(cmd_key, CMD_LIMIT, CMD_WINDOW)

    if not ok:
        await _block_user(user_id)
        metrics.spam_blocks.labels(kind="command").inc()
        return SpamResult.JUST_BLOCKED

    return SpamResult.ALLOWED


async def remaining_block(user_id: int) -> int:
    ttl = await r.ttl(_KEY_BLOCK.format(user_id=user_id))
    return max(0, ttl)