From 75f7b8ecf917eb7ec0e1729178ed3793bf2bf2a5 Mon Sep 17 00:00:00 2001 From: rambros Date: Tue, 24 Feb 2026 23:42:43 +0530 Subject: [PATCH] implement stoat token check --- src/core/discord_reader.py | 2 +- src/stoat/writer.py | 78 ++++++++++++++++++++++++++++++++------ src/ui/app.py | 19 ++++++---- 3 files changed, 80 insertions(+), 19 deletions(-) diff --git a/src/core/discord_reader.py b/src/core/discord_reader.py index cf12ba0..f0b0579 100644 --- a/src/core/discord_reader.py +++ b/src/core/discord_reader.py @@ -40,7 +40,7 @@ class DiscordReader: await temp_client.login(self.token) results["token"] = True if temp_client.user: - results["bot_name"] = temp_client.user.name + results["bot_name"] = temp_client.user.display_name guild = await temp_client.fetch_guild(self.server_id) if guild is not None: diff --git a/src/stoat/writer.py b/src/stoat/writer.py index 52a67e1..cf12d7d 100644 --- a/src/stoat/writer.py +++ b/src/stoat/writer.py @@ -1,4 +1,5 @@ import logging +import stoat from typing import Optional, List, Dict, Any logger = logging.getLogger(__name__) @@ -9,22 +10,77 @@ class StoatWriter: self.community_id = str(community_id) async def start(self): - logger.info("StoatWriter start (Not implemented)") + self.client = stoat.Client(token=self.token, bot=True) async def validate(self) -> dict: - return { - "token": True, - "community": True, - "bot_name": "Stoat Dummy", - "community_name": "Stoat Community Dummy", + results = { + "token": False, + "community": False, + "bot_name": "N/A", + "community_name": "N/A", "permissions": { - "manage_channels": True, - "manage_messages": True, - "manage_roles": True, - "manage_emojis_stickers": True, - "manage_webhooks": True + "manage_channels": False, + "manage_server": False, + "manage_permissions": False, + "manage_roles": False, + "manage_customization": False } } + + # Use a temporary client for validation + client = stoat.Client(token=self.token, bot=True) + try: + # Validate token by fetching current user + try: + current_user = await client.fetch_user("@me") + results["token"] = True + results["bot_name"] = current_user.display_name or current_user.name + except stoat.Unauthorized: + logger.error("Invalid Stoat token.") + return results + + # Validate server access + try: + server = await client.fetch_server(self.community_id) + results["community"] = True + results["community_name"] = server.name + + # Check permissions using effective server permissions for the bot + # Use current_user.id since @me might not be supported in all member endpoints + try: + me = await server.fetch_member(current_user.id) + # We use server.permissions_for(me) instead of me.server_permissions + # to avoid cache-related NoData exceptions. + # safe=False allows calculating even if some roles aren't in local cache. + perms = server.permissions_for(me, safe=False) + + results["permissions"] = { + "manage_channels": perms.manage_channels, + "manage_server": perms.manage_server, + "manage_permissions": perms.manage_permissions, + "manage_roles": perms.manage_roles, + "manage_customization": perms.manage_customization + } + except stoat.NotFound: + logger.error(f"Bot member {current_user.id} not found in Stoat server {self.community_id}.") + except stoat.Forbidden: + logger.error(f"Bot lacks permissions to fetch its own member data in Stoat server {self.community_id}.") + except Exception as e: + logger.error(f"Error fetching Stoat member permissions: {e}") + + except stoat.NotFound: + logger.error(f"Stoat server {self.community_id} not found.") + except stoat.Forbidden: + logger.error(f"Bot has no access to Stoat server {self.community_id}.") + except Exception as e: + logger.error(f"Error validating Stoat server: {e}") + + except Exception as e: + logger.error(f"Stoat validation failed: {str(e)}") + finally: + await client.close() + + return results async def get_channels(self) -> List[Dict[str, Any]]: return [] diff --git a/src/ui/app.py b/src/ui/app.py index b98cb78..a5f19d0 100644 --- a/src/ui/app.py +++ b/src/ui/app.py @@ -215,10 +215,15 @@ class MigrationCLI: if not all([d_intents.get("message_content"), d_perms.get("view_channel"), d_perms.get("read_message_history")]): self.permissions_complete = False - # Fluxer - f_perms = self.validation_results.get("fluxer_permissions", {}) - if not all(f_perms.values()) if f_perms else True: - self.permissions_complete = False + # Fluxer or Stoat + if self.target_platform == "fluxer": + f_perms = self.validation_results.get("fluxer_permissions", {}) + if not all(f_perms.values()) if f_perms else True: + self.permissions_complete = False + else: + s_perms = self.validation_results.get("stoat_permissions", {}) + if not all(s_perms.values()) if s_perms else True: + self.permissions_complete = False except Exception as e: console.print(f"[bold red]Validation system failure: {e}[/bold red]") @@ -250,7 +255,7 @@ class MigrationCLI: while True: console.print("") - console.print(Panel.fit("Fluxer Reaper", style="bold blue")) + console.print(Panel.fit(f"{self.target_platform.capitalize()} Reaper", style="bold blue")) d_name = self.validation_results.get("discord_server_name") d_display = f"[bold green]\"{d_name}\"[/bold green]" if d_name else ("[bold yellow]TIMEOUT ERROR[/bold yellow]" if self.validation_results.get("discord_timeout") else "[bold red]NOT SET UP[/bold red]") @@ -338,8 +343,8 @@ class MigrationCLI: s_perms = self.validation_results.get("stoat_permissions", {}) perm_table.add_row( "[bold #FF8C00]Stoat Bot[/bold #FF8C00]", - f"• [bold]Permissions:[/bold] {fmt('Manage Channels', s_perms.get('manage_channels'))}, {fmt('Manage Messages', s_perms.get('manage_messages'))},\n" - f" {fmt('Manage Roles', s_perms.get('manage_roles'))}, {fmt('Manage Emojis/Stickers', s_perms.get('manage_emojis_stickers'))}, {fmt('Manage Webhooks', s_perms.get('manage_webhooks'))}" + f"• [bold]Permissions:[/bold] {fmt('Manage Channel', s_perms.get('manage_channels'))}, {fmt('Manage Server', s_perms.get('manage_server'))},\n" + f" {fmt('Manage Permissions', s_perms.get('manage_permissions'))}, {fmt('Manage Roles', s_perms.get('manage_roles'))}, {fmt('Manage Customization', s_perms.get('manage_customization'))}" ) console.print("\n") # Add spacing before panel