Added trusted API route and bot checker

api/app/routers/v1/trusted.py

@@ -0,0 +1,24 @@
+from fastapi import APIRouter
+
+from app.database.models import Trusted
+
+router = APIRouter()
+
+
+@router.post("/", response_model=Trusted, status_code=201)
+async def create_trusted_user(trusted: Trusted) -> Trusted:
+    return await trusted.save()
+
+
+@router.get("/{uid}", response_model=list[Trusted])
+async def user_is_trusted(uid: int) -> Trusted:
+    is_trusted = await Trusted.objects.get(pk=uid)
+
+    return {"is_trusted": is_trusted}
+
+
+@router.delete("/{uid}", response_model=Trusted)
+async def remove_user_trust(uid: int) -> Trusted:
+    item_db = await Trusted.objects.get(pk=uid)
+
+    return {"deleted_rows": await item_db.delete()}

xythrion/bot/api.py

@@ -1,6 +1,14 @@
+from dataclasses import dataclass
+
 from httpx import AsyncClient, Response
 
 
+@dataclass
+class InternalAPIResponse:
+    status: int
+    data: dict
+
+
 class APIClient:
     def __init__(self, base_url: str) -> None:
         self.http_client = AsyncClient()
@@ -14,20 +22,18 @@ def full_url(self, partial_endpoint: str) -> str:
 
         return f"{self.base_url}{extra_path_sep}{partial_endpoint}"
 
-    async def request(self, method: str, partial_endpoint: str, **kwargs) -> dict:
+    async def request(self, method: str, partial_endpoint: str, **kwargs) -> InternalAPIResponse:
         r: Response = await self.http_client.request(
             method.upper(), self.full_url(partial_endpoint), **kwargs
         )
 
-        r.raise_for_status()
-
-        return r.json()
+        return InternalAPIResponse(r.status_code, r.json())
 
-    async def get(self, partial_endpoint: str, **kwargs) -> dict:
+    async def get(self, partial_endpoint: str, **kwargs) -> InternalAPIResponse:
         return await self.request("GET", partial_endpoint, **kwargs)
 
-    async def post(self, partial_endpoint: str, **kwargs) -> dict:
+    async def post(self, partial_endpoint: str, **kwargs) -> InternalAPIResponse:
         return await self.request("POST", partial_endpoint, **kwargs)
 
-    async def delete(self, partial_endpoint: str, **kwargs) -> dict:
+    async def delete(self, partial_endpoint: str, **kwargs) -> InternalAPIResponse:
         return await self.request("DELETE", partial_endpoint, **kwargs)

xythrion/bot/extensions/core/administration.py

@@ -1,8 +1,9 @@
-from discord.ext.commands import Cog, command, is_owner
+from discord.ext.commands import Cog, command
 from loguru import logger as log
 
 from bot.bot import Xythrion
 from bot.context import Context
+from bot.utils import is_trusted
 
 
 class Administration(Cog):
@@ -12,7 +13,7 @@ def __init__(self, bot: Xythrion) -> None:
         self.bot = bot
 
     @command(alias=("logout",))
-    @is_owner()
+    @is_trusted()
     async def shutdown(self, ctx: Context) -> None:
         """Shuts the bot down."""
         log.info("Logging out...")

xythrion/bot/extensions/core/eval.py

@@ -8,9 +8,10 @@
 from typing import Any
 
 import discord
-from discord.ext.commands import Cog, Context, group, is_owner
+from discord.ext.commands import Cog, Context, group
 
 from bot.bot import Xythrion
+from bot.utils import is_trusted
 
 
 def find_nth_occurrence(string: str, substring: str, n: int) -> int | None:
@@ -205,14 +206,14 @@ async def func():  # (None,) -> Any
         return None
 
     @group(name="internal", aliases=("int",))
-    @is_owner()
+    @is_trusted()
     async def internal_group(self, ctx: Context) -> None:
         """Internal commands. Top secret!"""
         if not ctx.invoked_subcommand:
             await ctx.send_help(ctx.command)
 
     @internal_group.command(name="eval", aliases=("e",))
-    @is_owner()
+    @is_trusted()
     async def eval(self, ctx: Context, *, code: str) -> None:
         """Run eval in a REPL-like format."""
         code = code.strip("`")

xythrion/bot/extensions/core/extensions.py

@@ -1,9 +1,9 @@
-from discord.ext.commands import Cog, ExtensionNotLoaded, group, is_owner
+from discord.ext.commands import Cog, ExtensionNotLoaded, group
 from loguru import logger as log
 
 from bot.bot import EXTENSIONS, Xythrion
 from bot.context import Context
-from bot.utils import Extension, codeblock
+from bot.utils import Extension, codeblock, is_trusted
 
 
 class Extensions(Cog):
@@ -13,7 +13,7 @@ def __init__(self, bot: Xythrion):
         self.bot = bot
 
     @group(aliases=("extensions", "e"))
-    @is_owner()
+    @is_trusted()
     async def extension(self, ctx: Context) -> None:
         """Extension group command."""
         if ctx.invoked_subcommand is None:

xythrion/bot/extensions/graph_generation/expressions.py

@@ -1,8 +1,8 @@
-from discord.ext.commands import Cog, group, is_owner
+from discord.ext.commands import Cog, group
 
 from bot.bot import Xythrion
 from bot.context import Context
-from bot.utils import remove_whitespace
+from bot.utils import is_trusted, remove_whitespace
 
 
 class GraphExpression(Cog):
@@ -12,14 +12,14 @@ def __init__(self, bot: Xythrion) -> None:
         self.bot = bot
 
     @group()
-    @is_owner()
+    @is_trusted()
     async def graph(self, ctx: Context) -> None:
         """Group command for graphing."""
         if ctx.invoked_subcommand is None:
             await ctx.send("Missing subcommand")
 
     @graph.command(aliases=("ex", "expr"))
-    @is_owner()
+    @is_trusted()
     async def expression(self, ctx: Context, *, expression: remove_whitespace) -> None:
         """Parse an expression into its components and graph it."""
         await ctx.send(expression)

xythrion/bot/utils/__init__.py

@@ -1,12 +1,14 @@
+from bot.utils.checks import is_trusted
 from bot.utils.converters import (
     Extension,
     convert_3d_tuples,
     remove_whitespace,
 )
-from bot.utils.formatting import and_join, codeblock, markdown_link
+from bot.utils.formatting import codeblock, markdown_link
 from bot.utils.gradients import gradient3
 
 __all__ = (
+    "is_trusted",
     "and_join",
     "codeblock",
     "markdown_link",

xythrion/bot/utils/checks.py

@@ -0,0 +1,18 @@
+from collections.abc import Callable
+
+from discord.ext.commands import check
+
+from bot.api import InternalAPIResponse
+from bot.context import Context
+
+
+def is_trusted() -> Callable:
+    async def predicate(ctx: Context) -> bool:
+        if await ctx.bot.is_owner(ctx.message.author):
+            return True
+
+        r: InternalAPIResponse = await ctx.bot.api.get(f"/v1/trusted/{ctx.message.author.id}")
+
+        return r.status == 200
+
+    return check(predicate)