[Core] Add more APIs for allowlists and blocklists (#5206)

* [Core] Blacklist api

* [Core] Use to_add, remove star from other method calls

* various touch ups

* style

* fix doc style

* [Core] Remove iterable import

* [Core] Update commands to use the blacklist/whitelist api

* Change signatures to not use `*args`

* Update all usage of private cache to the new public APIs

* Update the docstrings

* Update the usage in diagnoser

Co-authored-by: Kreusada <67752638+Kreusada@users.noreply.github.com>
Co-authored-by: jack1142 <6032823+jack1142@users.noreply.github.com>
This commit is contained in:
Just-Jojo 2021-09-06 12:38:07 -04:00 committed by GitHub
parent ed9bb77eec
commit 0dded8aa47
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 184 additions and 39 deletions

View File

@ -166,7 +166,7 @@ class DetailedGlobalCallOnceChecksMixin(IssueDiagnoserBase):
) )
async def _get_detailed_global_whitelist_blacklist_result(self, label: str) -> CheckResult: async def _get_detailed_global_whitelist_blacklist_result(self, label: str) -> CheckResult:
global_whitelist = await self.bot._whiteblacklist_cache.get_whitelist() global_whitelist = await self.bot.get_whitelist()
if global_whitelist: if global_whitelist:
return CheckResult( return CheckResult(
False, False,
@ -206,7 +206,7 @@ class DetailedGlobalCallOnceChecksMixin(IssueDiagnoserBase):
async def _get_detailed_local_whitelist_blacklist_result(self, label: str) -> CheckResult: async def _get_detailed_local_whitelist_blacklist_result(self, label: str) -> CheckResult:
# this method skips guild owner check as the earlier checks wouldn't fail # this method skips guild owner check as the earlier checks wouldn't fail
# if the user were guild owner # if the user were guild owner
guild_whitelist = await self.bot._whiteblacklist_cache.get_whitelist(self.guild) guild_whitelist = await self.bot.get_whitelist(self.guild)
if guild_whitelist: if guild_whitelist:
return CheckResult( return CheckResult(
False, False,
@ -227,7 +227,7 @@ class DetailedGlobalCallOnceChecksMixin(IssueDiagnoserBase):
) )
details = _("Local blocklist prevents the user from running this command.") details = _("Local blocklist prevents the user from running this command.")
guild_blacklist = await self.bot._whiteblacklist_cache.get_blacklist(self.guild) guild_blacklist = await self.bot.get_blacklist(self.guild)
ids = {role.id for role in self.author.roles if not role.is_default()} ids = {role.id for role in self.author.roles if not role.is_default()}
ids.add(self.author.id) ids.add(self.author.id)
intersection = ids & guild_blacklist intersection = ids & guild_blacklist

View File

@ -17,6 +17,7 @@ from typing import (
Optional, Optional,
Union, Union,
List, List,
Iterable,
Dict, Dict,
NoReturn, NoReturn,
Set, Set,
@ -26,6 +27,7 @@ from typing import (
Any, Any,
Literal, Literal,
MutableMapping, MutableMapping,
Set,
overload, overload,
) )
from types import MappingProxyType from types import MappingProxyType
@ -66,6 +68,7 @@ DataDeletionResults = namedtuple("DataDeletionResults", "failed_modules failed_c
PreInvokeCoroutine = Callable[[commands.Context], Awaitable[Any]] PreInvokeCoroutine = Callable[[commands.Context], Awaitable[Any]]
T_BIC = TypeVar("T_BIC", bound=PreInvokeCoroutine) T_BIC = TypeVar("T_BIC", bound=PreInvokeCoroutine)
UserOrRole = Union[int, discord.Role, discord.Member, discord.User]
def _is_submodule(parent, child): def _is_submodule(parent, child):
@ -523,6 +526,156 @@ class RedBase(
def max_messages(self) -> Optional[int]: def max_messages(self) -> Optional[int]:
return self._max_messages return self._max_messages
async def add_to_blacklist(
self, users_or_roles: Iterable[UserOrRole], *, guild: Optional[discord.Guild] = None
):
"""
Add users or roles to the global or local blocklist.
Parameters
----------
users_or_roles : Iterable[Union[int, discord.Role, discord.Member, discord.User]]
The items to add to the blocklist.
Roles and role IDs should only be passed when updating a local blocklist.
guild : Optional[discord.Guild]
The guild, whose local blocklist should be modified.
If not passed, the global blocklist will be modified.
Raises
------
TypeError
The values passed were not of the proper type.
"""
to_add: Set[int] = {getattr(uor, "id", uor) for uor in users_or_roles}
await self._whiteblacklist_cache.add_to_blacklist(guild, to_add)
async def remove_from_blacklist(
self, users_or_roles: Iterable[UserOrRole], *, guild: Optional[discord.Guild] = None
):
"""
Remove users or roles from the global or local blocklist.
Parameters
----------
users_or_roles : Iterable[Union[int, discord.Role, discord.Member, discord.User]]
The items to remove from the blocklist.
Roles and role IDs should only be passed when updating a local blocklist.
guild : Optional[discord.Guild]
The guild, whose local blocklist should be modified.
If not passed, the global blocklist will be modified.
Raises
------
TypeError
The values passed were not of the proper type.
"""
to_remove: Set[int] = {getattr(uor, "id", uor) for uor in users_or_roles}
await self._whiteblacklist_cache.remove_from_blacklist(guild, to_remove)
async def get_blacklist(self, guild: Optional[discord.Guild] = None) -> Set[int]:
"""
Get the global or local blocklist.
Parameters
----------
guild : Optional[discord.Guild]
The guild to get the local blocklist for.
If this is not passed, the global blocklist will be returned.
Returns
-------
Set[int]
The IDs of the blocked users/roles.
"""
return await self._whiteblacklist_cache.get_blacklist(guild)
async def clear_blacklist(self, guild: Optional[discord.Guild] = None):
"""
Clears the global or local blocklist.
Parameters
----------
guild : Optional[discord.Guild]
The guild, whose local blocklist should be cleared.
If not passed, the global blocklist will be cleared.
"""
await self._whiteblacklist_cache.clear_blacklist(guild)
async def add_to_whitelist(
self, users_or_roles: Iterable[UserOrRole], *, guild: Optional[discord.Guild] = None
):
"""
Add users or roles to the global or local allowlist.
Parameters
----------
users_or_roles : Iterable[Union[int, discord.Role, discord.Member, discord.User]]
The items to add to the allowlist.
Roles and role IDs should only be passed when updating a local allowlist.
guild : Optional[discord.Guild]
The guild, whose local allowlist should be modified.
If not passed, the global allowlist will be modified.
Raises
------
TypeError
The passed values were not of the proper type.
"""
to_add: Set[int] = {getattr(uor, "id", uor) for uor in users_or_roles}
await self._whiteblacklist_cache.add_to_whitelist(guild, to_add)
async def remove_from_whitelist(
self, users_or_roles: Iterable[UserOrRole], *, guild: Optional[discord.Guild] = None
):
"""
Remove users or roles from the global or local allowlist.
Parameters
----------
users_or_roles : Iterable[Union[int, discord.Role, discord.Member, discord.User]]
The items to remove from the allowlist.
Roles and role IDs should only be passed when updating a local allowlist.
guild : Optional[discord.Guild]
The guild, whose local allowlist should be modified.
If not passed, the global allowlist will be modified.
Raises
------
TypeError
The passed values were not of the proper type.
"""
to_remove: Set[int] = {getattr(uor, "id", uor) for uor in users_or_roles}
await self._whiteblacklist_cache.remove_from_whitelist(guild, to_remove)
async def get_whitelist(self, guild: Optional[discord.Guild] = None):
"""
Get the global or local allowlist.
Parameters
----------
guild : Optional[discord.Guild]
The guild to get the local allowlist for.
If this is not passed, the global allowlist will be returned.
Returns
-------
Set[int]
The IDs of the allowed users/roles.
"""
return await self._whiteblacklist_cache.get_whitelist(guild)
async def clear_whitelist(self, guild: Optional[discord.Guild] = None):
"""
Clears the global or local allowlist.
Parameters
----------
guild : Optional[discord.Guild]
The guild, whose local allowlist should be cleared.
If not passed, the global allowlist will be cleared.
"""
await self._whiteblacklist_cache.clear_whitelist(guild)
async def allowed_by_whitelist_blacklist( async def allowed_by_whitelist_blacklist(
self, self,
who: Optional[Union[discord.Member, discord.User]] = None, who: Optional[Union[discord.Member, discord.User]] = None,
@ -534,7 +687,7 @@ class RedBase(
) -> bool: ) -> bool:
""" """
This checks if a user or member is allowed to run things, This checks if a user or member is allowed to run things,
as considered by Red's whitelist and blacklist. as considered by Red's allowlist and blocklist.
If given a user object, this function will check the global lists If given a user object, this function will check the global lists
@ -614,13 +767,13 @@ class RedBase(
if await self.is_owner(who): if await self.is_owner(who):
return True return True
global_whitelist = await self._whiteblacklist_cache.get_whitelist() global_whitelist = await self.get_whitelist()
if global_whitelist: if global_whitelist:
if who.id not in global_whitelist: if who.id not in global_whitelist:
return False return False
else: else:
# blacklist is only used when whitelist doesn't exist. # blacklist is only used when whitelist doesn't exist.
global_blacklist = await self._whiteblacklist_cache.get_blacklist() global_blacklist = await self.get_blacklist()
if who.id in global_blacklist: if who.id in global_blacklist:
return False return False
@ -648,12 +801,12 @@ class RedBase(
# there is a silent failure potential, and role blacklist/whitelists will break. # there is a silent failure potential, and role blacklist/whitelists will break.
ids = {i for i in (who.id, *(getattr(who, "_roles", []))) if i != guild.id} ids = {i for i in (who.id, *(getattr(who, "_roles", []))) if i != guild.id}
guild_whitelist = await self._whiteblacklist_cache.get_whitelist(guild) guild_whitelist = await self.get_whitelist(guild)
if guild_whitelist: if guild_whitelist:
if ids.isdisjoint(guild_whitelist): if ids.isdisjoint(guild_whitelist):
return False return False
else: else:
guild_blacklist = await self._whiteblacklist_cache.get_blacklist(guild) guild_blacklist = await self.get_blacklist(guild)
if not ids.isdisjoint(guild_blacklist): if not ids.isdisjoint(guild_blacklist):
return False return False

View File

@ -3765,9 +3765,8 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Arguments:** **Arguments:**
- `<users...>` - The user or users to add to the allowlist. - `<users...>` - The user or users to add to the allowlist.
""" """
uids = {getattr(user, "id", user) for user in users} await self.bot.add_to_whitelist(users)
await self.bot._whiteblacklist_cache.add_to_whitelist(None, uids) if len(users) > 1:
if len(uids) > 1:
await ctx.send(_("Users have been added to the allowlist.")) await ctx.send(_("Users have been added to the allowlist."))
else: else:
await ctx.send(_("User has been added to the allowlist.")) await ctx.send(_("User has been added to the allowlist."))
@ -3812,9 +3811,8 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Arguments:** **Arguments:**
- `<users...>` - The user or users to remove from the allowlist. - `<users...>` - The user or users to remove from the allowlist.
""" """
uids = {getattr(user, "id", user) for user in users} await self.bot.remove_from_whitelist(users)
await self.bot._whiteblacklist_cache.remove_from_whitelist(None, uids) if len(users) > 1:
if len(uids) > 1:
await ctx.send(_("Users have been removed from the allowlist.")) await ctx.send(_("Users have been removed from the allowlist."))
else: else:
await ctx.send(_("User has been removed from the allowlist.")) await ctx.send(_("User has been removed from the allowlist."))
@ -3829,7 +3827,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]allowlist clear` - `[p]allowlist clear`
""" """
await self.bot._whiteblacklist_cache.clear_whitelist() await self.bot.clear_whitelist()
await ctx.send(_("Allowlist has been cleared.")) await ctx.send(_("Allowlist has been cleared."))
@commands.group(aliases=["blacklist", "denylist"]) @commands.group(aliases=["blacklist", "denylist"])
@ -3863,9 +3861,8 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
await ctx.send(_("You cannot add an owner to the blocklist!")) await ctx.send(_("You cannot add an owner to the blocklist!"))
return return
uids = {getattr(user, "id", user) for user in users} await self.bot.add_to_blacklist(users)
await self.bot._whiteblacklist_cache.add_to_blacklist(None, uids) if len(users) > 1:
if len(uids) > 1:
await ctx.send(_("Users have been added to the blocklist.")) await ctx.send(_("Users have been added to the blocklist."))
else: else:
await ctx.send(_("User has been added to the blocklist.")) await ctx.send(_("User has been added to the blocklist."))
@ -3878,7 +3875,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]blocklist list` - `[p]blocklist list`
""" """
curr_list = await self.bot._whiteblacklist_cache.get_blacklist(None) curr_list = await self.bot.get_blacklist()
if not curr_list: if not curr_list:
await ctx.send("Blocklist is empty.") await ctx.send("Blocklist is empty.")
@ -3908,9 +3905,8 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Arguments:** **Arguments:**
- `<users...>` - The user or users to remove from the blocklist. - `<users...>` - The user or users to remove from the blocklist.
""" """
uids = {getattr(user, "id", user) for user in users} await self.bot.remove_from_blacklist(users)
await self.bot._whiteblacklist_cache.remove_from_blacklist(None, uids) if len(users) > 1:
if len(uids) > 1:
await ctx.send(_("Users have been removed from the blocklist.")) await ctx.send(_("Users have been removed from the blocklist."))
else: else:
await ctx.send(_("User has been removed from the blocklist.")) await ctx.send(_("User has been removed from the blocklist."))
@ -3923,7 +3919,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]blocklist clear` - `[p]blocklist clear`
""" """
await self.bot._whiteblacklist_cache.clear_blacklist() await self.bot.clear_blacklist()
await ctx.send(_("Blocklist has been cleared.")) await ctx.send(_("Blocklist has been cleared."))
@commands.group(aliases=["localwhitelist"]) @commands.group(aliases=["localwhitelist"])
@ -3957,7 +3953,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
names = [getattr(u_or_r, "name", u_or_r) for u_or_r in users_or_roles] names = [getattr(u_or_r, "name", u_or_r) for u_or_r in users_or_roles]
uids = {getattr(u_or_r, "id", u_or_r) for u_or_r in users_or_roles} uids = {getattr(u_or_r, "id", u_or_r) for u_or_r in users_or_roles}
if not (ctx.guild.owner == ctx.author or await self.bot.is_owner(ctx.author)): if not (ctx.guild.owner == ctx.author or await self.bot.is_owner(ctx.author)):
current_whitelist = await self.bot._whiteblacklist_cache.get_whitelist(ctx.guild) current_whitelist = await self.bot.get_whitelist(ctx.guild)
theoretical_whitelist = current_whitelist.union(uids) theoretical_whitelist = current_whitelist.union(uids)
ids = {i for i in (ctx.author.id, *(getattr(ctx.author, "_roles", [])))} ids = {i for i in (ctx.author.id, *(getattr(ctx.author, "_roles", [])))}
if ids.isdisjoint(theoretical_whitelist): if ids.isdisjoint(theoretical_whitelist):
@ -3968,7 +3964,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
"please ensure to add yourself to the allowlist first." "please ensure to add yourself to the allowlist first."
) )
) )
await self.bot._whiteblacklist_cache.add_to_whitelist(ctx.guild, uids) await self.bot.add_to_whitelist(uids, guild=ctx.guild)
if len(uids) > 1: if len(uids) > 1:
await ctx.send(_("Users and/or roles have been added to the allowlist.")) await ctx.send(_("Users and/or roles have been added to the allowlist."))
@ -3983,7 +3979,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]localallowlist list` - `[p]localallowlist list`
""" """
curr_list = await self.bot._whiteblacklist_cache.get_whitelist(ctx.guild) curr_list = await self.bot.get_whitelist(ctx.guild)
if not curr_list: if not curr_list:
await ctx.send("Server allowlist is empty.") await ctx.send("Server allowlist is empty.")
@ -4021,7 +4017,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
names = [getattr(u_or_r, "name", u_or_r) for u_or_r in users_or_roles] names = [getattr(u_or_r, "name", u_or_r) for u_or_r in users_or_roles]
uids = {getattr(u_or_r, "id", u_or_r) for u_or_r in users_or_roles} uids = {getattr(u_or_r, "id", u_or_r) for u_or_r in users_or_roles}
if not (ctx.guild.owner == ctx.author or await self.bot.is_owner(ctx.author)): if not (ctx.guild.owner == ctx.author or await self.bot.is_owner(ctx.author)):
current_whitelist = await self.bot._whiteblacklist_cache.get_whitelist(ctx.guild) current_whitelist = await self.bot.get_whitelist(ctx.guild)
theoretical_whitelist = current_whitelist - uids theoretical_whitelist = current_whitelist - uids
ids = {i for i in (ctx.author.id, *(getattr(ctx.author, "_roles", [])))} ids = {i for i in (ctx.author.id, *(getattr(ctx.author, "_roles", [])))}
if theoretical_whitelist and ids.isdisjoint(theoretical_whitelist): if theoretical_whitelist and ids.isdisjoint(theoretical_whitelist):
@ -4031,7 +4027,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
"remove your ability to run commands." "remove your ability to run commands."
) )
) )
await self.bot._whiteblacklist_cache.remove_from_whitelist(ctx.guild, uids) await self.bot.remove_from_whitelist(uids, guild=ctx.guild)
if len(uids) > 1: if len(uids) > 1:
await ctx.send(_("Users and/or roles have been removed from the server allowlist.")) await ctx.send(_("Users and/or roles have been removed from the server allowlist."))
@ -4048,7 +4044,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]localallowlist clear` - `[p]localallowlist clear`
""" """
await self.bot._whiteblacklist_cache.clear_whitelist(ctx.guild) await self.bot.clear_whitelist(ctx.guild)
await ctx.send(_("Server allowlist has been cleared.")) await ctx.send(_("Server allowlist has been cleared."))
@commands.group(aliases=["localblacklist"]) @commands.group(aliases=["localblacklist"])
@ -4088,11 +4084,9 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
if await ctx.bot.is_owner(uid): if await ctx.bot.is_owner(uid):
await ctx.send(_("You cannot add a bot owner to the blocklist!")) await ctx.send(_("You cannot add a bot owner to the blocklist!"))
return return
names = [getattr(u_or_r, "name", u_or_r) for u_or_r in users_or_roles] await self.bot.add_to_blacklist(users_or_roles, guild=ctx.guild)
uids = {getattr(u_or_r, "id", u_or_r) for u_or_r in users_or_roles}
await self.bot._whiteblacklist_cache.add_to_blacklist(ctx.guild, uids)
if len(uids) > 1: if len(users_or_roles) > 1:
await ctx.send(_("Users and/or roles have been added from the server blocklist.")) await ctx.send(_("Users and/or roles have been added from the server blocklist."))
else: else:
await ctx.send(_("User or role has been added from the server blocklist.")) await ctx.send(_("User or role has been added from the server blocklist."))
@ -4105,7 +4099,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]localblocklist list` - `[p]localblocklist list`
""" """
curr_list = await self.bot._whiteblacklist_cache.get_blacklist(ctx.guild) curr_list = await self.bot.get_blacklist(ctx.guild)
if not curr_list: if not curr_list:
await ctx.send("Server blocklist is empty.") await ctx.send("Server blocklist is empty.")
@ -4138,11 +4132,9 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Arguments:** **Arguments:**
- `<users_or_roles...>` - The users or roles to remove from the local blocklist. - `<users_or_roles...>` - The users or roles to remove from the local blocklist.
""" """
names = [getattr(u_or_r, "name", u_or_r) for u_or_r in users_or_roles] await self.bot.remove_from_blacklist(users_or_roles, guild=ctx.guild)
uids = {getattr(u_or_r, "id", u_or_r) for u_or_r in users_or_roles}
await self.bot._whiteblacklist_cache.remove_from_blacklist(ctx.guild, uids)
if len(uids) > 1: if len(users_or_roles) > 1:
await ctx.send(_("Users and/or roles have been removed from the server blocklist.")) await ctx.send(_("Users and/or roles have been removed from the server blocklist."))
else: else:
await ctx.send(_("User or role has been removed from the server blocklist.")) await ctx.send(_("User or role has been removed from the server blocklist."))
@ -4157,7 +4149,7 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
**Example:** **Example:**
- `[p]blocklist clear` - `[p]blocklist clear`
""" """
await self.bot._whiteblacklist_cache.clear_blacklist(ctx.guild) await self.bot.clear_blacklist(ctx.guild)
await ctx.send(_("Server blocklist has been cleared.")) await ctx.send(_("Server blocklist has been cleared."))
@checks.guildowner_or_permissions(administrator=True) @checks.guildowner_or_permissions(administrator=True)