Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,17 @@ All notable changes to this project will be documented in this file.

The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [Unreleased]

### Changed

- `automod` improvements (#41):
- Implemented role-based (per-guild) permissions
- Added a new `log_message` action that suppresses pings by default
- Pings are now suppressed by default in error messages
- Added normalization to the `message_content_contains` condition
- Added more fields to some events for string formatting

## [0.13.0] - 2021-08-22

### Added
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,5 +18,7 @@ def get_target(self, event: AutomodEvent) -> Optional[Member]:
async def apply(self, event: AutomodEvent):
if member := self.get_target(event):
guild: Guild = member.guild
# TODO Warn about unresolved roles. #logging
roles = [guild.get_role(role_id) for role_id in self.roles]
roles = [role for role in roles if role]
await member.add_roles(roles)
Original file line number Diff line number Diff line change
Expand Up @@ -18,5 +18,7 @@ def get_target(self, event: AutomodEvent) -> Optional[Member]:
async def apply(self, event: AutomodEvent):
if member := self.get_target(event):
guild: Guild = member.guild
# TODO Warn about unresolved roles. #logging
roles = [guild.get_role(role_id) for role_id in self.roles]
roles = [role for role in roles if role]
await member.remove_roles(roles)
90 changes: 90 additions & 0 deletions commanderbot_ext/ext/automod/actions/log_message.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
from dataclasses import dataclass, field
from typing import Dict, Optional, Type, TypeVar

from discord import Color
from discord.abc import Messageable

from commanderbot_ext.ext.automod.automod_action import AutomodAction, AutomodActionBase
from commanderbot_ext.ext.automod.automod_event import AutomodEvent
from commanderbot_ext.lib import AllowedMentions, ChannelID, JsonObject, ValueFormatter
from commanderbot_ext.lib.utils import color_from_field_optional

ST = TypeVar("ST")


@dataclass
class LogMessage(AutomodActionBase):
"""
Send a log message, with pings disabled by default.

Attributes
----------
content
The content of the message to send.
channel
The channel to send the message in. Defaults to the channel in context.
emoji
The emoji used to represent the type of message.
color
The emoji used to represent the type of message.
fields
A custom set of fields to display as part of the message. The key should
correspond to an event field, and the value is the title to use for it.
allowed_mentions
The types of mentions allowed in the message. Unless otherwise specified, all
mentions will be suppressed.
"""

content: Optional[str] = None
channel: Optional[ChannelID] = None
emoji: Optional[str] = None
color: Optional[Color] = None
fields: Optional[Dict[str, str]] = None
allowed_mentions: Optional[AllowedMentions] = None

@classmethod
def from_data(cls: Type[ST], data: JsonObject) -> ST:
color = color_from_field_optional(data, "color")
allowed_mentions = AllowedMentions.from_field_optional(data, "allowed_mentions")
return cls(
description=data.get("description"),
content=data.get("content"),
channel=data.get("channel"),
emoji=data.get("emoji"),
color=color,
fields=data.get("fields"),
allowed_mentions=allowed_mentions,
)

async def resolve_channel(self, event: AutomodEvent) -> Optional[Messageable]:
if self.channel is not None:
return event.bot.get_channel(self.channel)
return event.channel

async def apply(self, event: AutomodEvent):
if channel := await self.resolve_channel(event):
parts = []
if self.emoji:
parts.append(self.emoji)
if self.content:
formatted_content = event.format_content(self.content)
parts.append(formatted_content)
if self.fields:
event_fields = event.get_fields()
field_parts = []
for field_key, field_title in self.fields.items():
field_value = event_fields.get(field_key)
if isinstance(field_value, ValueFormatter):
field_value_str = str(field_value)
else:
field_value_str = f"`{field_value}`"
field_parts.append(f"{field_title} {field_value_str}")
fields_str = ", ".join(field_parts)
parts.append(fields_str)
content = " ".join(parts)
allowed_mentions = self.allowed_mentions or AllowedMentions.none()
await channel.send(content, allowed_mentions=allowed_mentions)


def create_action(data: JsonObject) -> AutomodAction:
return LogMessage.from_data(data)
29 changes: 24 additions & 5 deletions commanderbot_ext/ext/automod/actions/send_message.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
from dataclasses import dataclass
from typing import Optional
from dataclasses import dataclass, field
from typing import Optional, Type, TypeVar

from discord.abc import Messageable

from commanderbot_ext.ext.automod.automod_action import AutomodAction, AutomodActionBase
from commanderbot_ext.ext.automod.automod_event import AutomodEvent
from commanderbot_ext.lib import ChannelID, JsonObject
from commanderbot_ext.lib import AllowedMentions, ChannelID, JsonObject

ST = TypeVar("ST")


@dataclass
Expand All @@ -19,11 +21,24 @@ class SendMessage(AutomodActionBase):
The content of the message to send.
channel
The channel to send the message in. Defaults to the channel in context.
allowed_mentions
The types of mentions allowed in the message. Unless otherwise specified, only
"everyone" mentions will be suppressed.
"""

content: str

channel: Optional[ChannelID] = None
allowed_mentions: Optional[AllowedMentions] = None

@classmethod
def from_data(cls: Type[ST], data: JsonObject) -> ST:
allowed_mentions = AllowedMentions.from_field_optional(data, "allowed_mentions")
return cls(
description=data.get("description"),
content=data.get("content"),
channel=data.get("channel"),
allowed_mentions=allowed_mentions,
)

async def resolve_channel(self, event: AutomodEvent) -> Optional[Messageable]:
if self.channel is not None:
Expand All @@ -33,7 +48,11 @@ async def resolve_channel(self, event: AutomodEvent) -> Optional[Messageable]:
async def apply(self, event: AutomodEvent):
if channel := await self.resolve_channel(event):
content = event.format_content(self.content)
await channel.send(content)
allowed_mentions = self.allowed_mentions or AllowedMentions.not_everyone()
await channel.send(
content,
allowed_mentions=allowed_mentions,
)


def create_action(data: JsonObject) -> AutomodAction:
Expand Down
67 changes: 65 additions & 2 deletions commanderbot_ext/ext/automod/automod_cog.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,12 @@
RawMessageUpdateEvent,
RawReactionActionEvent,
Reaction,
Role,
TextChannel,
User,
)
from discord.abc import Messageable
from discord.ext import commands
from discord.ext.commands import Bot, Cog, group

from commanderbot_ext.ext.automod.automod_data import AutomodData
Expand Down Expand Up @@ -53,6 +55,17 @@ def make_automod_store(bot: Bot, cog: Cog, options: AutomodOptions) -> AutomodSt
raise UnsupportedDatabaseOptions(db_options)


def check_can_run_automod():
async def predicate(ctx: GuildContext):
cog = ctx.cog
actor = ctx.author
if isinstance(cog, AutomodCog) and isinstance(actor, Member):
return await cog.state[ctx.guild].member_has_permission(actor)
return False

return commands.check(predicate)


class AutomodCog(Cog, name="commanderbot_ext.ext.automod"):
"""
Automate a variety of moderation tasks.
Expand Down Expand Up @@ -254,7 +267,11 @@ async def on_raw_reaction_remove(self, payload: RawReactionActionEvent):
aliases=["am"],
)
@checks.guild_only()
@checks.is_administrator()
@commands.check_any(
checks.is_administrator(),
check_can_run_automod(),
commands.is_owner(),
)
async def cmd_automod(self, ctx: GuildContext):
if not ctx.invoked_subcommand:
await ctx.send_help(self.cmd_automod)
Expand All @@ -269,17 +286,26 @@ async def cmd_automod_options(self, ctx: GuildContext):
if not ctx.invoked_subcommand:
await ctx.send_help(self.cmd_automod_options)

# @@ automod options log

@cmd_automod_options.group(
name="log",
brief="Configure the default logging behaviour.",
)
async def cmd_automod_options_log(self, ctx: GuildContext):
if not ctx.invoked_subcommand:
if ctx.subcommand_passed:
await ctx.send_help(self.cmd_automod_options)
await ctx.send_help(self.cmd_automod_options_log)
else:
await self.state[ctx.guild].show_default_log_options(ctx)

@cmd_automod_options_log.command(
name="show",
brief="Show the default logging behaviour.",
)
async def cmd_automod_options_log_show(self, ctx: GuildContext):
await self.state[ctx.guild].show_default_log_options(ctx)

@cmd_automod_options_log.command(
name="set",
brief="Set the default logging behaviour.",
Expand Down Expand Up @@ -307,6 +333,43 @@ async def cmd_automod_options_log_set(
async def cmd_automod_options_log_remove(self, ctx: GuildContext):
await self.state[ctx.guild].remove_default_log_options(ctx)

# @@ automod options permit

# NOTE Only guild admins and bot owners can manage permitted roles.

@cmd_automod_options.group(
name="permit",
brief="Configure the set of roles permitted to manage automod.",
)
@checks.is_guild_admin_or_bot_owner()
async def cmd_automod_options_permit(self, ctx: GuildContext):
if not ctx.invoked_subcommand:
if ctx.subcommand_passed:
await ctx.send_help(self.cmd_automod_options_permit)
else:
await self.state[ctx.guild].show_permitted_roles(ctx)

@cmd_automod_options_permit.command(
name="show",
brief="Show the roles permitted to manage automod.",
)
async def cmd_automod_options_permit_show(self, ctx: GuildContext):
await self.state[ctx.guild].show_permitted_roles(ctx)

@cmd_automod_options_permit.command(
name="set",
brief="Set the roles permitted to manage automod.",
)
async def cmd_automod_options_permit_set(self, ctx: GuildContext, *roles: Role):
await self.state[ctx.guild].set_permitted_roles(ctx, *roles)

@cmd_automod_options_permit.command(
name="clear",
brief="Clear all roles permitted to manage automod.",
)
async def cmd_automod_options_permit_clear(self, ctx: GuildContext):
await self.state[ctx.guild].clear_permitted_roles(ctx)

# @@ automod rules

@cmd_automod.group(
Expand Down
49 changes: 42 additions & 7 deletions commanderbot_ext/ext/automod/automod_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,12 @@
from datetime import datetime
from typing import AsyncIterable, DefaultDict, Dict, Iterable, Optional, Set, Type

from discord import Guild
from discord import Guild, Member

from commanderbot_ext.ext.automod.automod_event import AutomodEvent
from commanderbot_ext.ext.automod.automod_log_options import AutomodLogOptions
from commanderbot_ext.ext.automod.automod_rule import AutomodRule
from commanderbot_ext.lib import GuildID, JsonObject, ResponsiveException
from commanderbot_ext.lib import GuildID, JsonObject, ResponsiveException, RoleSet
from commanderbot_ext.lib.json import to_data
from commanderbot_ext.lib.utils import dict_without_ellipsis

Expand Down Expand Up @@ -51,20 +51,27 @@ def __init__(self, names: Set[str]):

@dataclass
class AutomodGuildData:
# Default logging configuration for this guild.
default_log_options: Optional[AutomodLogOptions] = None

# Index rules by name for fast look-up in commands.
# Roles that are permitted to manage the extension within this guild.
permitted_roles: Optional[RoleSet] = None

# Index rules by name for faster look-up in commands.
rules: Dict[str, AutomodRule] = field(init=False, default_factory=dict)

# Index rules by event type for faster initial access.
# Group rules by event type for faster look-up during event dispatch.
rules_by_event_type: RulesByEventType = field(
init=False, default_factory=lambda: defaultdict(lambda: set())
)

@staticmethod
def from_data(data: JsonObject) -> AutomodGuildData:
default_log_options = AutomodLogOptions.from_field_optional(data, "log")
permitted_roles = RoleSet.from_field_optional(data, "permitted_roles")
guild_data = AutomodGuildData(
default_log_options=AutomodLogOptions.from_field_optional(data, "log"),
default_log_options=default_log_options,
permitted_roles=permitted_roles,
)
for rule_data in data.get("rules", []):
rule = AutomodRule.from_data(rule_data)
Expand All @@ -74,15 +81,29 @@ def from_data(data: JsonObject) -> AutomodGuildData:
def to_data(self) -> JsonObject:
return dict_without_ellipsis(
log=self.default_log_options or ...,
permitted_roles=self.permitted_roles or ...,
rules=list(self.rules.values()) or ...,
)

def set_default_log_options(
self, log_options: Optional[AutomodLogOptions]
) -> Optional[AutomodLogOptions]:
old_log_options = self.default_log_options
old_value = self.default_log_options
self.default_log_options = log_options
return old_log_options
return old_value

def set_permitted_roles(
self, permitted_roles: Optional[RoleSet]
) -> Optional[RoleSet]:
old_value = self.permitted_roles
self.permitted_roles = permitted_roles
return old_value

def member_has_permission(self, member: Member) -> bool:
if self.permitted_roles is None:
return False
has_permission = self.permitted_roles.member_has_some(member)
return has_permission

def all_rules(self) -> Iterable[AutomodRule]:
yield from self.rules.values()
Expand Down Expand Up @@ -224,6 +245,20 @@ async def set_default_log_options(
) -> Optional[AutomodLogOptions]:
return self.guilds[guild.id].set_default_log_options(log_options)

# @implements AutomodStore
async def get_permitted_roles(self, guild: Guild) -> Optional[RoleSet]:
return self.guilds[guild.id].permitted_roles

# @implements AutomodStore
async def set_permitted_roles(
self, guild: Guild, permitted_roles: Optional[RoleSet]
) -> Optional[RoleSet]:
return self.guilds[guild.id].set_permitted_roles(permitted_roles)

# @implements AutomodStore
async def member_has_permission(self, guild: Guild, member: Member) -> bool:
return self.guilds[guild.id].member_has_permission(member)

# @implements AutomodStore
async def all_rules(self, guild: Guild) -> AsyncIterable[AutomodRule]:
for rule in self.guilds[guild.id].all_rules():
Expand Down
Loading