Allow central storage of API keys (#2389)

This creates a central location to store external API tokens that can be used between cogs without requiring each cog to be loaded for it to work.

A new set option for `[p]set api` is created to assist in forming bot readable API token locations.

This also updates the Streams cog to utilize the central database.

Tokens are moved from the old data locations in core cogs on load.

redbot/cogs/image/__init__.py

@@ -1,6 +1,7 @@
 from .image import Image
 
 
-def setup(bot):
-    n = Image(bot)
-    bot.add_cog(n)
+async def setup(bot):
+    cog = Image(bot)
+    await cog.initialize()
+    bot.add_cog(cog)

redbot/cogs/image/image.py

@@ -27,6 +27,13 @@ class Image(commands.Cog):
     def __unload(self):
         self.session.detach()
 
+    async def initialize(self) -> None:
+        """Move the API keys from cog stored config to core bot config if they exist."""
+        imgur_token = await self.settings.imgur_client_id()
+        if imgur_token is not None and "imgur" not in await self.bot.db.api_tokens():
+            await self.bot.db.api_tokens.set_raw("imgur", value={"client_id": imgur_token})
+            await self.settings.imgur_client_id.clear()
+
     @commands.group(name="imgur")
     async def _imgur(self, ctx):
         """Retrieve pictures from Imgur.
@@ -43,7 +50,7 @@ class Image(commands.Cog):
         """
         url = self.imgur_base_url + "gallery/search/time/all/0"
         params = {"q": term}
-        imgur_client_id = await self.settings.imgur_client_id()
+        imgur_client_id = await ctx.bot.db.api_tokens.get_raw("imgur", default=None)
         if not imgur_client_id:
             await ctx.send(
                 _(
@@ -51,7 +58,7 @@ class Image(commands.Cog):
                 ).format(prefix=ctx.prefix)
             )
             return
-        headers = {"Authorization": "Client-ID {}".format(imgur_client_id)}
+        headers = {"Authorization": "Client-ID {}".format(imgur_client_id["client_id"])}
         async with self.session.get(url, headers=headers, params=params) as search_get:
             data = await search_get.json()
 
@@ -96,7 +103,7 @@ class Image(commands.Cog):
             await ctx.send_help()
             return
 
-        imgur_client_id = await self.settings.imgur_client_id()
+        imgur_client_id = await ctx.bot.db.api_tokens.get_raw("imgur", default=None)
         if not imgur_client_id:
             await ctx.send(
                 _(
@@ -106,7 +113,7 @@ class Image(commands.Cog):
             return
 
         links = []
-        headers = {"Authorization": "Client-ID {}".format(imgur_client_id)}
+        headers = {"Authorization": "Client-ID {}".format(imgur_client_id["client_id"])}
         url = self.imgur_base_url + "gallery/r/{}/{}/{}/0".format(subreddit, sort, window)
 
         async with self.session.get(url, headers=headers) as sub_get:
@@ -130,22 +137,24 @@ class Image(commands.Cog):
 
     @checks.is_owner()
     @commands.command()
-    async def imgurcreds(self, ctx, imgur_client_id: str):
-        """Set the Imgur Client ID.
+    async def imgurcreds(self, ctx):
+        """Explain how to set imgur API tokens"""
 
-        To get an Imgur Client ID:
-        1. Login to an Imgur account.
-        2. Visit [this](https://api.imgur.com/oauth2/addclient) page
-        3. Enter a name for your application
-        4. Select *Anonymous usage without user authorization* for the auth type
-        5. Set the authorization callback URL to `https://localhost`
-        6. Leave the app website blank
-        7. Enter a valid email address and a description
-        8. Check the captcha box and click next
-        9. Your Client ID will be on the next page.
-        """
-        await self.settings.imgur_client_id.set(imgur_client_id)
-        await ctx.send(_("The Imgur Client ID has been set!"))
+        message = _(
+            "To get an Imgur Client ID:\n"
+            "1. Login to an Imgur account.\n"
+            "2. Visit [this](https://api.imgur.com/oauth2/addclient) page\n"
+            "3. Enter a name for your application\n"
+            "4. Select *Anonymous usage without user authorization* for the auth type\n"
+            "5. Set the authorization callback URL to `https://localhost`\n"
+            "6. Leave the app website blank\n"
+            "7. Enter a valid email address and a description\n"
+            "8. Check the captcha box and click next\n"
+            "9. Your Client ID will be on the next page.\n"
+            "10. do `{prefix}set api imgur client_id,your_client_id`\n"
+        ).format(prefix=ctx.prefix)
+
+        await ctx.maybe_send_embed(message)
 
     @commands.guild_only()
     @commands.command()

redbot/cogs/streams/streams.py

@@ -71,22 +71,36 @@ class Streams(commands.Cog):
 
     async def initialize(self) -> None:
         """Should be called straight after cog instantiation."""
+        await self.move_api_keys()
         self.streams = await self.load_streams()
         self.communities = await self.load_communities()
 
         self.task = self.bot.loop.create_task(self._stream_alerts())
 
+    async def move_api_keys(self):
+        """Move the API keys from cog stored config to core bot config if they exist."""
+        tokens = await self.db.tokens()
+        youtube = await self.bot.db.api_tokens.get_raw("youtube", default={})
+        twitch = await self.bot.db.api_tokens.get_raw("twitch", default={})
+        for token_type, token in tokens.items():
+            if token_type == "YoutubeStream" and "api_key" not in youtube:
+                await self.bot.db.api_tokens.set_raw("youtube", value={"api_key": token})
+            if token_type == "TwitchStream" and "client_id" not in twitch:
+                # Don't need to check Community since they're set the same
+                await self.bot.db.api_tokens.set_raw("twitch", value={"client_id": token})
+        await self.db.tokens.clear()
+
     @commands.command()
     async def twitch(self, ctx: commands.Context, channel_name: str):
         """Check if a Twitch channel is live."""
-        token = await self.db.tokens.get_raw(TwitchStream.__name__, default=None)
+        token = await self.bot.db.api_tokens.get_raw("twitch", default={"client_id": None})
         stream = TwitchStream(name=channel_name, token=token)
         await self.check_online(ctx, stream)
 
     @commands.command()
     async def youtube(self, ctx: commands.Context, channel_id_or_name: str):
         """Check if a YouTube channel is live."""
-        apikey = await self.db.tokens.get_raw(YoutubeStream.__name__, default=None)
+        apikey = await self.bot.db.api_tokens.get_raw("youtube", default={"api_key": None})
         is_name = self.check_name_or_id(channel_id_or_name)
         if is_name:
             stream = YoutubeStream(name=channel_id_or_name, token=apikey)
@@ -253,7 +267,7 @@ class Streams(commands.Cog):
     async def stream_alert(self, ctx: commands.Context, _class, channel_name):
         stream = self.get_stream(_class, channel_name)
         if not stream:
-            token = await self.db.tokens.get_raw(_class.__name__, default=None)
+            token = await self.bot.db.api_tokens.get_raw(_class.token_name, default=None)
             is_yt = _class.__name__ == "YoutubeStream"
             if is_yt and not self.check_name_or_id(channel_name):
                 stream = _class(id=channel_name, token=token)
@@ -292,7 +306,7 @@ class Streams(commands.Cog):
     async def community_alert(self, ctx: commands.Context, _class, community_name):
         community = self.get_community(_class, community_name)
         if not community:
-            token = await self.db.tokens.get_raw(_class.__name__, default=None)
+            token = await self.bot.db.api_tokens.get_raw(_class.token_name, default=None)
             community = _class(name=community_name, token=token)
             try:
                 await community.get_community_streams()
@@ -325,36 +339,42 @@ class Streams(commands.Cog):
 
     @streamset.command()
     @checks.is_owner()
-    async def twitchtoken(self, ctx: commands.Context, token: str):
-        """Set the Client ID for Twitch.
+    async def twitchtoken(self, ctx: commands.Context):
+        """Explain how to set the twitch token"""
 
-        To do this, follow these steps:
-        1. Go to this page: https://dev.twitch.tv/dashboard/apps.
-        2. Click *Register Your Application*
-        3. Enter a name, set the OAuth Redirect URI to `http://localhost`, and
-           select an Application Category of your choosing.
-        4. Click *Register*, and on the following page, copy the Client ID.
-        5. Paste the Client ID into this command. Done!
-        """
-        await self.db.tokens.set_raw("TwitchStream", value=token)
-        await self.db.tokens.set_raw("TwitchCommunity", value=token)
-        await ctx.send(_("Twitch token set."))
+        message = _(
+            "To set the twitch API tokens, follow these steps:\n"
+            "1. Go to this page: https://dev.twitch.tv/dashboard/apps.\n"
+            "2. Click *Register Your Application*\n"
+            "3. Enter a name, set the OAuth Redirect URI to `http://localhost`, and \n"
+            "select an Application Category of your choosing."
+            "4. Click *Register*, and on the following page, copy the Client ID.\n"
+            "5. do `{prefix}set api twitch client_id,your_client_id`\n\n"
+            "Note: These tokens are sensitive and should only be used in a private channel\n"
+            "or in DM with the bot.)\n"
+        ).format(prefix=ctx.prefix)
+
+        await ctx.maybe_send_embed(message)
 
     @streamset.command()
     @checks.is_owner()
-    async def youtubekey(self, ctx: commands.Context, key: str):
-        """Set the API key for YouTube.
+    async def youtubekey(self, ctx: commands.Context):
+        """Explain how to set the YouTube token"""
 
-        To get one, do the following:
-        1. Create a project (see https://support.google.com/googleapi/answer/6251787 for details)
-        2. Enable the YouTube Data API v3 (see https://support.google.com/googleapi/answer/6158841
-        for instructions)
-        3. Set up your API key (see https://support.google.com/googleapi/answer/6158862 for
-        instructions)
-        4. Copy your API key and paste it into this command. Done!
-        """
-        await self.db.tokens.set_raw("YoutubeStream", value=key)
-        await ctx.send(_("YouTube key set."))
+        message = _(
+            "To get one, do the following:\n"
+            "1. Create a project\n"
+            "(see https://support.google.com/googleapi/answer/6251787 for details)\n"
+            "2. Enable the YouTube Data API v3 \n"
+            "(see https://support.google.com/googleapi/answer/6158841for instructions)\n"
+            "3. Set up your API key \n"
+            "(see https://support.google.com/googleapi/answer/6158862 for instructions)\n"
+            "4. Copy your API key and do `{prefix}set api youtube api_key,your_api_key`\n\n"
+            "Note: These tokens are sensitive and should only be used in a private channel\n"
+            "or in DM with the bot.\n"
+        ).format(prefix=ctx.prefix)
+
+        await ctx.maybe_send_embed(message)
 
     @streamset.group()
     @commands.guild_only()
@@ -656,7 +676,7 @@ class Streams(commands.Cog):
                         pass
                     else:
                         raw_stream["_messages_cache"].append(msg)
-            token = await self.db.tokens.get_raw(_class.__name__, default=None)
+            token = await self.bot.db.api_tokens.get_raw(_class.token_name, default=None)
             if token is not None:
                 raw_stream["token"] = token
             streams.append(_class(**raw_stream))
@@ -681,7 +701,7 @@ class Streams(commands.Cog):
                         pass
                     else:
                         raw_community["_messages_cache"].append(msg)
-            token = await self.db.tokens.get_raw(_class.__name__, default=None)
+            token = await self.bot.db.api_tokens.get_raw(_class.token_name, default=None)
             communities.append(_class(token=token, **raw_community))
 
         # issue 1191 extended resolution: Remove this after suitable period

redbot/cogs/streams/streamtypes.py

@@ -9,6 +9,7 @@ from .errors import (
 )
 from random import choice, sample
 from string import ascii_letters
+from typing import ClassVar, Optional
 import discord
 import aiohttp
 import json
@@ -30,6 +31,9 @@ def rnd(url):
 
 
 class TwitchCommunity:
+
+    token_name = "twitch"
+
     def __init__(self, **kwargs):
         self.name = kwargs.pop("name")
         self.id = kwargs.pop("id", None)
@@ -39,7 +43,10 @@ class TwitchCommunity:
         self.type = self.__class__.__name__
 
     async def get_community_id(self):
-        headers = {"Accept": "application/vnd.twitchtv.v5+json", "Client-ID": str(self._token)}
+        headers = {
+            "Accept": "application/vnd.twitchtv.v5+json",
+            "Client-ID": str(self._token["client_id"]),
+        }
         params = {"name": self.name}
         async with aiohttp.ClientSession() as session:
             async with session.get(
@@ -61,7 +68,10 @@ class TwitchCommunity:
                 self.id = await self.get_community_id()
             except CommunityNotFound:
                 raise
-        headers = {"Accept": "application/vnd.twitchtv.v5+json", "Client-ID": str(self._token)}
+        headers = {
+            "Accept": "application/vnd.twitchtv.v5+json",
+            "Client-ID": str(self._token["client_id"]),
+        }
         params = {"community_id": self.id, "limit": 100}
         url = TWITCH_BASE_URL + "/kraken/streams"
         async with aiohttp.ClientSession() as session:
@@ -80,7 +90,10 @@ class TwitchCommunity:
             raise APIError()
 
     async def make_embed(self, streams: list) -> discord.Embed:
-        headers = {"Accept": "application/vnd.twitchtv.v5+json", "Client-ID": str(self._token)}
+        headers = {
+            "Accept": "application/vnd.twitchtv.v5+json",
+            "Client-ID": str(self._token["client_id"]),
+        }
         async with aiohttp.ClientSession() as session:
             async with session.get(
                 "{}/{}".format(TWITCH_COMMUNITIES_ENDPOINT, self.id), headers=headers
@@ -118,6 +131,9 @@ class TwitchCommunity:
 
 
 class Stream:
+
+    token_name: ClassVar[Optional[str]] = None
+
     def __init__(self, **kwargs):
         self.name = kwargs.pop("name", None)
         self.channels = kwargs.pop("channels", [])
@@ -146,6 +162,9 @@ class Stream:
 
 
 class YoutubeStream(Stream):
+
+    token_name = "youtube"
+
     def __init__(self, **kwargs):
         self.id = kwargs.pop("id", None)
         self._token = kwargs.pop("token", None)
@@ -162,7 +181,7 @@ class YoutubeStream(Stream):
 
         url = YOUTUBE_SEARCH_ENDPOINT
         params = {
-            "key": self._token,
+            "key": self._token["api_key"],
             "part": "snippet",
             "channelId": self.id,
             "type": "video",
@@ -175,7 +194,7 @@ class YoutubeStream(Stream):
             raise OfflineStream()
         elif "items" in data:
             vid_id = data["items"][0]["id"]["videoId"]
-            params = {"key": self._token, "id": vid_id, "part": "snippet"}
+            params = {"key": self._token["api_key"], "id": vid_id, "part": "snippet"}
             async with aiohttp.ClientSession() as session:
                 async with session.get(YOUTUBE_VIDEOS_ENDPOINT, params=params) as r:
                     data = await r.json()
@@ -202,7 +221,7 @@ class YoutubeStream(Stream):
 
     async def _fetch_channel_resource(self, resource: str):
 
-        params = {"key": self._token, "part": resource}
+        params = {"key": self._token["api_key"], "part": resource}
         if resource == "id":
             params["forUsername"] = self.name
         else:
@@ -229,6 +248,9 @@ class YoutubeStream(Stream):
 
 
 class TwitchStream(Stream):
+
+    token_name = "twitch"
+
     def __init__(self, **kwargs):
         self.id = kwargs.pop("id", None)
         self._token = kwargs.pop("token", None)
@@ -239,7 +261,10 @@ class TwitchStream(Stream):
             self.id = await self.fetch_id()
 
         url = TWITCH_STREAMS_ENDPOINT + self.id
-        header = {"Client-ID": str(self._token), "Accept": "application/vnd.twitchtv.v5+json"}
+        header = {
+            "Client-ID": str(self._token["client_id"]),
+            "Accept": "application/vnd.twitchtv.v5+json",
+        }
 
         async with aiohttp.ClientSession() as session:
             async with session.get(url, headers=header) as r:
@@ -260,7 +285,10 @@ class TwitchStream(Stream):
             raise APIError()
 
     async def fetch_id(self):
-        header = {"Client-ID": str(self._token), "Accept": "application/vnd.twitchtv.v5+json"}
+        header = {
+            "Client-ID": str(self._token["client_id"]),
+            "Accept": "application/vnd.twitchtv.v5+json",
+        }
         url = TWITCH_ID_ENDPOINT + self.name
 
         async with aiohttp.ClientSession() as session:
@@ -303,6 +331,9 @@ class TwitchStream(Stream):
 
 
 class HitboxStream(Stream):
+
+    token_name = None  # This streaming services don't currently require an API key
+
     async def is_online(self):
         url = "https://api.hitbox.tv/media/live/" + self.name
 
@@ -340,6 +371,9 @@ class HitboxStream(Stream):
 
 
 class MixerStream(Stream):
+
+    token_name = None  # This streaming services don't currently require an API key
+
     async def is_online(self):
         url = "https://mixer.com/api/v1/channels/" + self.name
 
@@ -381,6 +415,9 @@ class MixerStream(Stream):
 
 
 class PicartoStream(Stream):
+
+    token_name = None  # This streaming services don't currently require an API key
+
     async def is_online(self):
         url = "https://api.picarto.tv/v1/channel/name/" + self.name
 

redbot/core/bot.py

@@ -56,6 +56,7 @@ class RedBase(commands.GroupMixin, commands.bot.BotBase, RPCMixin):
             help__tagline="",
             disabled_commands=[],
             disabled_command_msg="That command is disabled.",
+            api_tokens={},
         )
 
         self.db.register_guild(

redbot/core/commands/converter.py

@@ -39,3 +39,29 @@ class GuildConverter(discord.Guild):
             raise BadArgument(_('Server "{name}" not found.').format(name=argument))
 
         return ret
+
+
+class APIToken(discord.ext.commands.Converter):
+    """Converts to a `dict` object.
+
+    This will parse the input argument separating the key value pairs into a 
+    format to be used for the core bots API token storage.
+    
+    This will split the argument by eiher `;` or `,` and return a dict
+    to be stored. Since all API's are different and have different naming convention,
+    this leaves the owness on the cog creator to clearly define how to setup the correct
+    credential names for their cogs.
+    """
+
+    async def convert(self, ctx, argument) -> dict:
+        bot = ctx.bot
+        result = {}
+        match = re.split(r";|,", argument)
+        # provide two options to split incase for whatever reason one is part of the api key we're using
+        if len(match) > 1:
+            result[match[0]] = "".join(r for r in match[1:])
+        else:
+            raise BadArgument(_("The provided tokens are not in a valid format."))
+        if not result:
+            raise BadArgument(_("The provided tokens are not in a valid format."))
+        return result

redbot/core/core_commands.py

@@ -1036,6 +1036,25 @@ class Core(commands.Cog, CoreLogic):
         else:
             await ctx.bot.send(_("Characters must be fewer than 1024."))
 
+    @_set.command()
+    @checks.is_owner()
+    async def api(self, ctx: commands.Context, service: str, *tokens: commands.converter.APIToken):
+        """Set various external API tokens.
+        
+        This setting will be asked for by some 3rd party cogs and some core cogs.
+
+        To add the keys provide the service name and the tokens as a comma separated
+        list of key,values as described by the cog requesting this command.
+
+        Note: API tokens are sensitive and should only be used in a private channel
+        or in DM with the bot.
+        """
+        if ctx.channel.permissions_for(ctx.me).manage_messages:
+            await ctx.message.delete()
+        entry = {k: v for t in tokens for k, v in t.items()}
+        await ctx.bot.db.api_tokens.set_raw(service, value=entry)
+        await ctx.send(_("`{service}` API tokens have been set.").format(service=service))
+
     @commands.group()
     @checks.is_owner()
     async def helpset(self, ctx: commands.Context):