diff options
Diffstat (limited to 'modules')
| -rw-r--r-- | modules/channel_op.py | 379 | ||||
| -rw-r--r-- | modules/highlight_spam.py | 31 |
2 files changed, 140 insertions, 270 deletions
diff --git a/modules/channel_op.py b/modules/channel_op.py index 151dc12a..719b0f50 100644 --- a/modules/channel_op.py +++ b/modules/channel_op.py @@ -5,17 +5,8 @@ from src import ModuleManager, utils -class UserNotFoundException(Exception): - pass -class InvalidTimeoutException(Exception): - pass +KICK_REASON = "your behavior is not conducive to the desired environment" -@utils.export("channelset", utils.IntSetting("highlight-spam-threshold", - "Set the number of nicknames in a message that qualifies as spam")) -@utils.export("channelset", utils.BoolSetting("highlight-spam-protection", - "Enable/Disable highlight spam protection")) -@utils.export("channelset", utils.BoolSetting("highlight-spam-ban", - "Enable/Disable banning highlight spammers instead of just kicking")) @utils.export("channelset", utils.Setting("ban-format", "Set ban format ($n = nick, $u = username, $h = hostname)", example="*!$u@$h")) @@ -23,293 +14,141 @@ class InvalidTimeoutException(Exception): ["qmode", "insp", "unreal", "none"], "Set this server's method of muting users")) class Module(ModuleManager.BaseModule): - _name = "ChanOp" + def _parse_time(self, args, min_args): + if args[0][0] == "+": + if len(args[1:]) < min_args: + raise utils.EventError("Not enough arguments") + time = utils.from_pretty_time(args[0][1:]) + if time == None: + raise utils.EventError("Invalid timeframe") + return time, args[1:] + return None, args - @utils.hook("timer.unban") - def _timer_unban(self, event): - server = self.bot.get_server_by_id(event["server_id"]) - if server and event["channel_name"] in server.channels: - channel = server.channels.get(event["channel_name"]) - channel.send_unban(event["hostmask"]) - - def _kick(self, server, channel, nickname, reason): - target_user = server.get_user(nickname) - if channel.has_user(target_user): - channel.send_kick(nickname, reason) + def _kick(self, server, channel, target_nickname, reason): + target_user = server.get_user(target_nickname, create=False) + if target_user and channel.has_user(target_user): + reason = " ".join(reason) or KICK_REASON + channel.send_kick(target_user.nickname, reason) else: - raise UserNotFoundException("That user is not in this channel") - def _kick_command(self, event, channel, args_split): - target = args_split[0] - reason = " ".join(args_split[1:]) or None - - try: - self._kick(event["server"], channel, target, reason) - except UserNotFoundException as e: - event["stderr"].write(str(e)) - - @utils.hook("received.command.kick", private_only=True, min_args=2) - def private_kick(self, event): - """ - :help: Kick a user from the current channel - :usage: <nickname> [reason] - :prefix: Kick - """ - channel = event["server"].channels.get(event["args_split"][0]) - - event["check_assert"](utils.Check("channel-access", channel, "kick")) - - self._kick_command(event, channel, event["args_split"][1:]) + raise utils.EventError("No such user") - @utils.hook("received.command.k", alias_of="kick") - @utils.hook("received.command.kick", channel_only=True, min_args=1) + @utils.hook("received.command.kick") + @utils.hook("received.command.k", alias_of="k") + @utils.kwarg("min_args", 1) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "kick") + @utils.kwarg("usage", "<nickname> [reason]") def kick(self, event): - """ - :help: Kick a user from the current channel - :usage: <nickname> [reason] - :require_mode: o - :require_access: kick - :prefix: Kick - """ - self._kick_command(event, event["target"], event["args_split"]) + self._kick(event["server"], event["target"], event["args_split"][0], + event["args_split"][1:]) - def _ban_format(self, user, s): + def _format_hostmask(self, user, s): return s.replace("$n", user.nickname).replace("$u", user.username ).replace("$h", user.hostname) - def _ban_user(self, channel, ban, user): - if channel.has_user(user): - format = channel.get_setting("ban-format", "*!$u@$h") - hostmask_split = format.split("$$") - hostmask_split = [self._ban_format(user, s) for s in hostmask_split] - hostmask = "".join(hostmask_split) - if ban: - channel.send_ban(hostmask) - else: - channel.send_unban(hostmask) - return hostmask - else: - raise UserNotFoundException("That user is not in this channel") + def _get_hostmask(self, channel, user): + format = channel.get_setting("ban-format", "*!$u@$h") + hostmask_split = [ + self._format_hostmask(user, s) for s in format.split("$$")] + return "$".join(hostmask_split) - def _ban(self, server, channel, ban, target): - target_user = server.get_user(target) - if channel.has_user(target_user): - return self._ban_user(channel, ban, target_user) + def _ban(self, server, channel, target, allow_hostmask, time, add): + hostmask = None + target_user = server.get_user(target, create=False) + if target_user and channel.has_user(target_user): + hostmask = self._get_hostmask(channel, target_user) else: - if ban: - channel.send_ban(target) - else: - channel.send_unban(target) - return target - - @utils.hook("received.command.ban", private_only=True, min_args=2) - def private_ban(self, event): - """ - :help: Ban a user/hostmask from the current channel - :usage: <channel> <nickname/hostmask> - """ - channel = event["server"].channels.get(event["args_split"][0]) - - event["check_assert"](utils.Check("channel-access", channel, "ban")) - - self._ban(event["server"], channel, True, event["args_split"][1]) - @utils.hook("received.command.ban", channel_only=True, min_args=1) - def ban(self, event): - """ - :help: Ban a user/hostmask from the current channel - :usage: <nickname/hostmask> - :require_mode: o - :require_access: ban - """ - self._ban(event["server"], event["target"], True, - event["args_split"][0]) - - def _temp_ban(self, event, accept_hostmask): - timeout = utils.from_pretty_time(event["args_split"][1]) - if not timeout: - raise InvalidTimeoutException( - "Please provided a valid time above 0 seconds") - - if accept_hostmask: - hostmask = self._ban(event["server"], event["target"], True, - event["args_split"][0]) + if not allow_hostmask: + raise utils.EventError("No such user") + hostmask = target + if not add: + channel.send_unban(hostmask) else: - hostmask = self._ban_user(event["target"], True, - event["server"].get_user(event["args_split"][0])) - - self.timers.add_persistent("unban", timeout, - server_id=event["server"].id, - channel_name=event["target"].name, hostmask=hostmask) - - @utils.hook("received.command.tb", alias_of="tempban") - @utils.hook("received.command.tempban", channel_only=True, min_args=2) - def temp_ban(self, event): - """ - :help: Temporarily ban someone from the current channel - :usage: <nickname/hostmask> <time> [reason] - :require_mode: o - :require_access: ban - :prefix: Tempban - """ - try: - self._temp_ban(event, True) - except InvalidTimeoutException as e: - event["stderr"].write(str(e)) + channel.send_ban(hostmask) - @utils.hook("received.command.tkb", alias_of="tempkickban") - @utils.hook("received.command.tempkickban", channel_only=True, - min_args=2) - def temp_kick_ban(self, event): - """ - :help: Temporarily kick and ban someone from the current channel - :usage: <nickname> <time> [reason] - :require_mode: o - :require_access: kickban - :prefix: TKB - """ - reason = " ".join(event["args_split"][2:]) or None - try: - self._temp_ban(event, False) - self._kick(event["server"], event["target"], event["args_split"][0], - reason) - except InvalidTimeoutException as e: - event["stderr"].write(str(e)) - except UserNotFoundException as e: - event["stderr"].write(str(e)) + if not time == None: + self.timers.add_persistent("unban", time, server_id=server.id, + channel_name=channel.name, hostmask=hostmask) - @utils.hook("received.command.unban", private_only=True, min_args=2) - def private_unban(self, event): - """ - :help: Unban a user/hostmask from the current channel - :usage: <channel> <nickname/hostmask> - """ - channel = event["server"].channels.get(event["args_split"][0]) - - event["check_assert"](utils.Check("channel-access", channel, "ban")) + @utils.hook("timer.unban") + def _timer_unban(self, event): + server = self.bot.get_server_by_id(event["server_id"]) + if server and event["channel_name"] in server.channels: + channel = server.channels.get(event["channel_name"]) + channel.send_unban(event["hostmask"]) - self._ban(event["server"], channel, False, event["args_split"][1]) + @utils.hook("received.command.ban") + @utils.kwarg("min_args", 1) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "ban") + @utils.kwarg("usage", "[+time] <target>") + def ban(self, event): + time, args = self._parse_time(event["args_split"], 1) + self._ban(event["server"], event["target"], args[0], True, time, True) - @utils.hook("received.command.unban", channel_only=True, min_args=1) + @utils.hook("received.command.unban") + @utils.kwarg("min_args", 1) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "ban") + @utils.kwarg("usage", "<target>") def unban(self, event): - """ - :help: Unban a user/hostmask from the current channel - :usage: <nickname/hostmask> - :require_mode: o - :require_access: ban - """ - self._ban(event["server"], event["target"], False, - event["args_split"][0]) + self._ban(event["server"], event["target"], event["args_split"][0], + True, None, False) + @utils.hook("received.command.kickban") @utils.hook("received.command.kb", alias_of="kickban") - @utils.hook("received.command.kickban", channel_only=True, min_args=1) + @utils.kwarg("min_args", 1) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "kickban") + @utils.kwarg("usage", "[+time] <nickname> [reason]") def kickban(self, event): - """ - :help: Kick and ban a user from the current channel - :usage: <nickname> [reason] - :require_mode: o - :require_access: kickban - :prefix: Kickban - """ - target = event["args_split"][0] - reason = " ".join(event["args_split"][1:]) or None - try: - self._ban(event["server"], event["target"], True, target) - self._kick(event["server"], event["target"], target, reason) - except UserNotFoundException as e: - event["stderr"].write(str(e)) + time, args = self._parse_time(event["args_split"], 1) + self._ban(event["server"], event["target"], args[0], False, time) + self._kick(event["server"], event["target"], args[0], args[1:]) - @utils.hook("received.command.op", channel_only=True) + @utils.hook("received.command.op") + @utils.hook("received.command.deop") + @utils.kwarg("channel_only", True) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "op") + @utils.kwarg("usage", "[nickname]") def op(self, event): - """ - :help: Op a user in the current channel - :usage: [nickname] - :require_mode: o - :require_access: op - """ - target = event["user"].nickname if not event["args_split"] else event[ - "args_split"][0] - event["target"].send_mode("+o", [target]) - @utils.hook("received.command.deop", channel_only=True) - def deop(self, event): - """ - :help: Remove op from a user in the current channel - :usage: [nickname] - :require_mode: o - :require_access: op - """ - target = event["user"].nickname if not event["args_split"] else event[ - "args_split"][0] - event["target"].send_mode("-o", [target]) + add = event.name == "received.command.op" + target = event["args_split"][0] if event["args"] else event[ + "user"].nickname + event["target"].send_mode("+o" if add else "-o", target) - @utils.hook("received.command.voice", channel_only=True) - def voice(self, event): - """ - :help: Voice a user in the current channel - :usage: [nickname] - :require_mode: o - :require_access: voice - """ - target = event["user"].nickname if not event["args_split"] else event[ - "args_split"][0] - event["target"].send_mode("+v", [target]) - @utils.hook("received.command.devoice", channel_only=True) - def devoice(self, event): - """ - :help: Remove voice from a user in the current channel - :usage: [nickname] - :require_mode: o - :require_access: voice - """ - target = event["user"].nickname if not event["args_split"] else event[ - "args_split"][0] - event["target"].send_mode("-v", [target]) + @utils.hook("received.command.voice") + @utils.hook("received.command.devoice") + @utils.kwarg("channel_only", True) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "voice") + @utils.kwarg("usage", "[nickname]") + def op(self, event): + add = event.name == "received.command.voice" + target = event["args_split"][0] if event["args"] else event[ + "user"].nickname + event["target"].send_mode("+v" if add else "-v", target) - @utils.hook("received.command.topic", min_args=1, channel_only=True, - remove_empty=False) + @utils.hook("received.command.topic") + @utils.kwarg("min_args", 1) + @utils.kwarg("channel_only", True) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "topic") + @utils.kwarg("remove_empty", False) + @utils.kwarg("usage", "<topic>") def topic(self, event): - """ - :help: Set the topic in the current channel - :usage: <topic> - :require_mode: o - :require_access: topic - """ event["target"].send_topic(event["args"]) - @utils.hook("received.command.tappend", min_args=1, channel_only=True, - remove_empty=False) + + @utils.hook("received.command.tappend") + @utils.kwarg("min_args", 1) + @utils.kwarg("channel_only", True) + @utils.kwarg("require_mode", "o") + @utils.kwarg("require_access", "topic") + @utils.kwarg("remove_empty", False) + @utils.kwarg("usage", "<topic>") def tappend(self, event): - """ - :help: Append to the topic in the current channel - :usage: <topic> - :require_mode: o - :require_access: topic - """ event["target"].send_topic(event["target"].topic + event["args"]) - @utils.hook("received.message.channel") - def highlight_spam(self, event): - if event["channel"].get_setting("highlight-spam-protection", False): - nicknames = list(map(lambda user: user.nickname, - event["channel"].users)) + [event["server"].nickname] - - highlights = set(nicknames) & set(event["message_split"]) - if len(highlights) > 1 and len(highlights) >= event["channel" - ].get_setting("highlight-spam-threshold", 10): - has_mode = event["channel"].mode_or_above(event["user"], "v") - should_ban = event["channel"].get_setting("highlight-spam-ban", - False) - if not has_mode: - if should_ban: - event["channel"].send_ban("*!%s@%s" % ( - event["user"].username, event["user"].hostname)) - event["channel"].send_kick(event["user"].nickname, - "highlight spam detected") - - @utils.hook("received.command.leave", channel_only=True) - def leave(self, event): - """ - :help: Part me from the current channel - :require_mode: o - """ - event["target"].send_part() - def _mute_method(self, server, user): mask = "*!*@%s" % user.hostname mute_method = server.get_setting("mute-method", "qmode").lower() diff --git a/modules/highlight_spam.py b/modules/highlight_spam.py new file mode 100644 index 00000000..8d02ee87 --- /dev/null +++ b/modules/highlight_spam.py @@ -0,0 +1,31 @@ +#--depends-on config + +from src import ModuleManager, utils + +@utils.export("channelset", utils.IntSetting("highlight-spam-threshold", + "Set the number of nicknames in a message that qualifies as spam")) +@utils.export("channelset", utils.BoolSetting("highlight-spam-protection", + "Enable/Disable highlight spam protection")) +@utils.export("channelset", utils.BoolSetting("highlight-spam-ban", + "Enable/Disable banning highlight spammers instead of just kicking")) +class Module(ModuleManager.BaseModule): + @utils.hook("received.message.channel") + def highlight_spam(self, event): + if event["channel"].get_setting("highlight-spam-protection", False): + nicknames = list(map(lambda user: user.nickname, + event["channel"].users)) + + highlights = set(nicknames) & set(event["message_split"]) + print(highlights) + if len(highlights) > 1 and len(highlights) >= event["channel" + ].get_setting("highlight-spam-threshold", 10): + has_mode = event["channel"].mode_or_above(event["user"], "v") + should_ban = event["channel"].get_setting("highlight-spam-ban", + False) + if not has_mode: + if should_ban: + event["channel"].send_ban("*!%s@%s" % ( + event["user"].username, event["user"].hostname)) + event["channel"].send_kick(event["user"].nickname, + "highlight spam detected") + |