path: root/osterman/protect.py

"""
Sopel plugin: osterman - channel protection core

Automatic enforcement (all configurable at runtime via !set):
  - Flood control          flood_threshold, flood_window_sec
  - Caps filter            caps_filter, caps_percent, caps_min_len
  - Repeat filter          repeat_filter, repeat_threshold, repeat_window_sec
  - Content filter         regex list via !filter
  - Bad word list          word list via !badword
  - Clone detection        clone_limit
  - Join flood             join_flood_filter, join_flood_count, join_flood_window
  - Nick flood             nick_flood_filter, nick_flood_count, nick_flood_window
  - Persistent ban reapply
  - Blacklist              global instant-ban on join
  - Auto-mode              auto +o/+h/+v on join (per channel)
  - Takeover mitigation    re-op ACL nicks that get deopped
  - Greet                  greet

Degrades gracefully by privilege level:
  op     - full: kick, ban, re-op, flood ban
  halfop - partial: kick only (no bans, no re-op)
  none   - passive: last_seen tracking only

Config is per-channel with global defaults as fallback.
  Global:      acl, whitelist, blacklist, config defaults
  Per-channel: config overrides, bans, filters, badwords, exceptions, auto-modes
"""

import fnmatch
import json
import logging
import os
import re
import threading
import time

from sopel import plugin
from sopel.config.types import FilenameAttribute, StaticSection

log = logging.getLogger(__name__)

_DB_DEFAULTS = {
    "config": {
        "flood_threshold":    5,
        "flood_window_sec":   10,
        "caps_filter":        1,
        "caps_percent":       70,
        "caps_min_len":       10,
        "repeat_filter":      1,
        "repeat_threshold":   3,
        "repeat_window_sec":  30,
        "clone_limit":        2,
        "join_flood_filter":  1,
        "join_flood_count":   5,
        "join_flood_window":  10,
        "nick_flood_filter":  1,
        "nick_flood_count":   3,
        "nick_flood_window":  60,
        "badword_action":     "kick",
        "tempban_max_min":    60,
        "idle_warn_min":      45,
        "idle_kick_min":      60,
        "log_limit":          6,
        "greet":              1,
    },
    "acl":        [],
    "whitelist":  ["*!*@services.dal.net"],
    "blacklist":  [],
    "exceptions": [],
    "channels":   {},
}


class OstermanSection(StaticSection):
    db_path = FilenameAttribute("db_path", relative=False,
                                default="/var/osterman/db.json")


def _load_db(path):
    if os.path.exists(path):
        with open(path) as f:
            data = json.load(f)
        data.setdefault("channels", {})
        data.setdefault("blacklist", [])
        return data
    db = {
        k: (dict(v) if isinstance(v, dict) else list(v))
        for k, v in _DB_DEFAULTS.items()
    }
    _save_db(path, db)
    return db


def _save_db(path, db):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        json.dump(db, f, indent=2)


def setup(bot):
    bot.config.define_section("osterman", OstermanSection, validate=False)
    path = bot.config.osterman.db_path

    bot.memory["os_db_path"]    = path
    bot.memory["os_lock"]       = threading.Lock()
    bot.memory["os_flood"]      = {}
    bot.memory["os_repeat"]     = {}
    bot.memory["os_join_flood"] = {}
    bot.memory["os_nick_flood"] = {}
    bot.memory["os_last"]       = {}

    try:
        bot.memory["os_db"] = _load_db(path)
    except Exception as exc:
        log.error("osterman: failed to load db from %s: %s - using defaults", path, exc)
        bot.memory["os_db"] = {
            k: (dict(v) if isinstance(v, dict) else list(v))
            for k, v in _DB_DEFAULTS.items()
        }


# --- DB helpers ---

def _global_db(bot):
    return bot.memory["os_db"]


def _chan_db(bot, channel):
    return bot.memory["os_db"]["channels"].setdefault(channel.lower(), {})


def _cfg(bot, channel):
    global_cfg  = bot.memory["os_db"].get("config", {})
    channel_cfg = _chan_db(bot, channel).get("config", {})
    return {**global_cfg, **channel_cfg}


def _save(bot):
    _save_db(bot.memory["os_db_path"], bot.memory["os_db"])


# --- Privilege helpers ---

def _hostmask(trigger):
    return f"{trigger.nick}!{trigger.user}@{trigger.host}"


def _bot_privs(bot, channel):
    chan = bot.channels.get(channel)
    if not chan:
        return 0
    return chan.privileges.get(bot.nick, 0)


def _bot_has_op(bot, channel):
    return _bot_privs(bot, channel) >= plugin.OP


def _bot_has_halfop(bot, channel):
    return _bot_privs(bot, channel) >= plugin.HALFOP


# --- Policy helpers ---

def _is_whitelisted(bot, hostmask):
    return any(fnmatch.fnmatch(hostmask, p)
               for p in bot.memory["os_db"].get("whitelist", []))


def _is_blacklisted(bot, hostmask):
    return any(fnmatch.fnmatch(hostmask, p)
               for p in bot.memory["os_db"].get("blacklist", []))


def _is_excepted(bot, hostmask, channel=None):
    # Check per-channel exceptions
    if channel:
        per_chan = _chan_db(bot, channel).get("exceptions", [])
        if any(fnmatch.fnmatch(hostmask, p) for p in per_chan):
            return True
    # Legacy: check global exceptions list for backward compat
    global_exc = bot.memory["os_db"].get("exceptions", [])
    return any(fnmatch.fnmatch(hostmask, p) for p in global_exc)


def _is_exempt(bot, hostmask, channel=None):
    return _is_whitelisted(bot, hostmask) or _is_excepted(bot, hostmask, channel)


def _is_banned(bot, hostmask, channel):
    global_bans  = bot.memory["os_db"].get("bans", [])
    channel_bans = _chan_db(bot, channel).get("bans", [])
    return any(fnmatch.fnmatch(hostmask, b["mask"])
               for b in global_bans + channel_bans)


def _matches_list(hostmask, entries):
    """Return True if hostmask matches any nick or hostmask pattern in entries."""
    for entry in entries:
        if "!" in entry or "@" in entry or "*" in entry or "?" in entry:
            if fnmatch.fnmatch(hostmask, entry):
                return True
        else:
            nick = hostmask.split("!")[0]
            if nick.lower() == entry.lower():
                return True
    return False


# --- Event handlers ---

@plugin.event("JOIN")
@plugin.require_chanmsg
def on_bot_join(bot, trigger):
    if trigger.nick != bot.nick:
        return
    channel = str(trigger.sender)
    if channel.lower() not in bot.memory["os_db"].get("channels", {}):
        _chan_db(bot, channel)
        _save(bot)


@plugin.event("JOIN")
@plugin.require_chanmsg
def on_join(bot, trigger):
    if trigger.nick == bot.nick:
        return

    channel    = str(trigger.sender)
    hm         = _hostmask(trigger)
    has_op     = _bot_has_op(bot, channel)
    has_halfop = _bot_has_halfop(bot, channel)

    # Blacklist: global instant-ban, checked before whitelist
    if _is_blacklisted(bot, hm):
        if has_op:
            bot.write(["MODE", channel, "+b", hm])
            bot.write(["KICK", channel, trigger.nick, "blacklisted"])
        elif has_halfop:
            bot.write(["KICK", channel, trigger.nick, "blacklisted"])
        return

    if _is_exempt(bot, hm, channel):
        # Still apply auto-modes for whitelisted/excepted users
        _apply_auto_modes(bot, trigger, channel, hm, has_op, has_halfop)
        return

    if _is_banned(bot, hm, channel):
        if has_op:
            bot.write(["MODE", channel, "+b", hm])
            bot.write(["KICK", channel, trigger.nick, "banned"])
        elif has_halfop:
            bot.write(["KICK", channel, trigger.nick, "banned"])
        return

    if not has_halfop:
        return

    cfg = _cfg(bot, channel)

    # Clone detection
    limit = cfg.get("clone_limit", 2)
    host  = trigger.host
    chan  = bot.channels.get(channel)
    if chan:
        clones = [u for u in chan.users
                  if u != trigger.nick and chan.users[u].host == host]
        if len(clones) >= limit:
            bot.write(["KICK", channel, trigger.nick, f"clone limit ({limit}) exceeded"])
            return

    # Join flood detection
    if cfg.get("join_flood_filter", 1):
        jf_count  = cfg.get("join_flood_count",  5)
        jf_window = cfg.get("join_flood_window", 10)
        now       = time.time()
        jf_key = (channel, host)
        times  = bot.memory["os_join_flood"].setdefault(jf_key, [])
        times.append(now)
        bot.memory["os_join_flood"][jf_key] = [t for t in times if now - t <= jf_window]
        if len(bot.memory["os_join_flood"][jf_key]) >= jf_count:
            mask = f"*!*@{host}"
            bot.write(["KICK", channel, trigger.nick, "join flood"])
            if has_op:
                bot.write(["MODE", channel, "+b", mask])
            bot.memory["os_join_flood"][jf_key] = []
            return

    # Auto-modes applied after all checks pass
    _apply_auto_modes(bot, trigger, channel, hm, has_op, has_halfop)


def _apply_auto_modes(bot, trigger, channel, hm, has_op, has_halfop):
    cdb = _chan_db(bot, channel)

    if has_op:
        autoop = cdb.get("autoop", [])
        if _matches_list(hm, autoop):
            bot.write(["MODE", channel, "+o", trigger.nick])
            return

        autohalfop = cdb.get("autohalfop", [])
        if _matches_list(hm, autohalfop):
            bot.write(["MODE", channel, "+h", trigger.nick])
            return

    if has_halfop or has_op:
        autovoice = cdb.get("autovoice", [])
        if _matches_list(hm, autovoice):
            bot.write(["MODE", channel, "+v", trigger.nick])


@plugin.event("MODE")
def on_mode(bot, trigger):
    if not trigger.args or len(trigger.args) < 2:
        return

    channel  = trigger.args[0]
    mode_str = trigger.args[1]
    targets  = list(trigger.args[2:])

    if channel not in bot.channels:
        return
    if not _bot_has_op(bot, channel):
        return

    acl    = _global_db(bot).get("acl", [])
    adding = True
    t_idx  = 0

    for ch in mode_str:
        if ch == "+":
            adding = True
        elif ch == "-":
            adding = False
        elif ch == "o":
            if t_idx < len(targets):
                target = targets[t_idx]
                t_idx += 1
                if not adding and target in acl:
                    bot.write(["MODE", channel, "+o", target])


@plugin.event("NICK")
def on_nick_flood(bot, trigger):
    if trigger.nick == bot.nick:
        return

    hm = _hostmask(trigger)
    if _is_exempt(bot, hm):
        return

    now      = time.time()
    host_key = f"{trigger.user}@{trigger.host}"

    global_cfg = _global_db(bot).get("config", {})
    if not global_cfg.get("nick_flood_filter", 1):
        return

    nf_count  = global_cfg.get("nick_flood_count",  3)
    nf_window = global_cfg.get("nick_flood_window", 60)

    times = bot.memory["os_nick_flood"].setdefault(host_key, [])
    times.append(now)
    bot.memory["os_nick_flood"][host_key] = [t for t in times if now - t <= nf_window]

    if len(bot.memory["os_nick_flood"][host_key]) >= nf_count:
        new_nick = trigger.args[0] if trigger.args else None
        if not new_nick:
            return
        for chan_name in list(bot.channels):
            if new_nick in bot.channels[chan_name].users:
                if _bot_has_halfop(bot, chan_name):
                    bot.write(["KICK", chan_name, new_nick, "nick flood"])
                    if _bot_has_op(bot, chan_name):
                        bot.write(["MODE", chan_name, "+b", hm])
        bot.memory["os_nick_flood"][host_key] = []


@plugin.rule(".*")
@plugin.require_chanmsg
def on_message(bot, trigger):
    nick = trigger.nick
    if nick == bot.nick:
        return

    channel = str(trigger.sender)
    hm      = _hostmask(trigger)
    now     = time.time()

    bot.memory["os_last"][(channel, nick)] = now

    if _is_exempt(bot, hm, channel):
        return

    has_op     = _bot_has_op(bot, channel)
    has_halfop = _bot_has_halfop(bot, channel)

    if not has_halfop:
        return

    with bot.memory["os_lock"]:
        cfg  = _cfg(bot, channel)
        cdb  = _chan_db(bot, channel)
        text = trigger.group(0) or ""

        # Regex content filter
        for pattern in cdb.get("filters", []):
            try:
                if re.search(pattern, text, re.IGNORECASE):
                    bot.write(["KICK", channel, nick, "content filter"])
                    return
            except re.error:
                pass

        # Bad word list
        text_lower = text.lower()
        for word in cdb.get("badwords", []):
            if word in text_lower:
                action = cfg.get("badword_action", "kick")
                if action == "warn":
                    bot.notice("watch your language", nick)
                else:
                    bot.write(["KICK", channel, nick, "bad word"])
                return

        # Caps filter
        if cfg.get("caps_filter", 1):
            letters = [c for c in text if c.isalpha()]
            if len(letters) >= cfg.get("caps_min_len", 10):
                upper_pct = sum(1 for c in letters if c.isupper()) / len(letters) * 100
                if upper_pct >= cfg.get("caps_percent", 70):
                    bot.write(["KICK", channel, nick, "caps"])
                    return

        # Repeat filter
        if cfg.get("repeat_filter", 1):
            rep_threshold = cfg.get("repeat_threshold",  3)
            rep_window    = cfg.get("repeat_window_sec", 30)
            msg_hash      = hash(text.strip().lower())
            history       = bot.memory["os_repeat"].setdefault((channel, nick), [])
            history.append((msg_hash, now))
            bot.memory["os_repeat"][(channel, nick)] = [
                (h, t) for h, t in history if now - t <= rep_window
            ]
            same = sum(1 for h, _ in bot.memory["os_repeat"][(channel, nick)]
                       if h == msg_hash)
            if same >= rep_threshold:
                bot.write(["KICK", channel, nick, "repeat"])
                bot.memory["os_repeat"][(channel, nick)] = []
                return

        # Flood control
        threshold = cfg.get("flood_threshold",  5)
        window    = cfg.get("flood_window_sec", 10)
        times     = bot.memory["os_flood"].setdefault((channel, nick), [])
        times.append(now)
        bot.memory["os_flood"][(channel, nick)] = [
            t for t in times if now - t <= window
        ]
        if len(bot.memory["os_flood"][(channel, nick)]) > threshold:
            bot.write(["KICK", channel, nick, "flood"])
            if has_op:
                bot.write(["MODE", channel, "+b", hm])
            bot.memory["os_flood"][(channel, nick)] = []


@plugin.event("PART")
@plugin.event("QUIT")
@plugin.event("NICK")
def on_gone(bot, trigger):
    nick     = trigger.nick
    host_key = f"{trigger.user}@{trigger.host}"

    if trigger.event == "PART":
        channel = str(trigger.sender)
        for cache in ("os_last", "os_flood", "os_repeat"):
            bot.memory.get(cache, {}).pop((channel, nick), None)
    else:
        for cache in ("os_last", "os_flood", "os_repeat"):
            mem  = bot.memory.get(cache, {})
            keys = [k for k in mem if isinstance(k, tuple) and k[1] == nick]
            for k in keys:
                del mem[k]

    bot.memory.get("os_nick_flood", {}).pop(host_key, None)