""" 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)