import asyncio import hashlib import hmac import json import logging import random import sys import aiohttp import discord from aiohttp import web from discord.ext import commands # Set up logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger("yasbbot") intents = discord.Intents.default() intents.members = True # Enable member intents intents.message_content = True # Add message content intent bot = commands.Bot(command_prefix="!", intents=intents) # Kofi webhook URL async def kofi_webhook(request): try: data = await request.post() kofi_data = json.loads(data.get("data", "{}")) # Get relevant fields is_public = kofi_data.get("is_public", True) is_subscription = kofi_data.get("is_subscription_payment", False) name = kofi_data.get("from_name", "Someone") # Determine display name display_name = name if is_public and name else "Someone" # Choose message type if is_subscription: msg = f"{display_name} just made a monthly donation on Ko-fi! 🥳" else: msg = f"{display_name} just made a donation on Ko-fi! 🥳" # Send to your channel channel = bot.get_channel(1353495378368008264) if channel: await channel.send(msg) return web.Response(text="OK") except Exception as e: logger.error(f"Error in Ko-fi webhook: {e}") return web.Response(status=500, text="Error") async def verify_github_signature(secret: str, body: bytes, signature_header: str) -> bool: """Verify GitHub HMAC SHA-256 signature (X-Hub-Signature-256). If secret is empty/None, verification is skipped and returns True. """ if not secret: # No secret configured; skip verification return True if not signature_header: logger.warning("No X-Hub-Signature-256 header present") return False try: sha_name, signature = signature_header.split("=", 1) except ValueError: logger.warning("Malformed X-Hub-Signature-256 header") return False if sha_name != "sha256": logger.warning(f"Unsupported hash algorithm: {sha_name}") return False mac = hmac.new(secret.encode(), msg=body, digestmod=hashlib.sha256) expected = mac.hexdigest() # Use hmac.compare_digest for timing-attack resistant comparison return hmac.compare_digest(expected, signature) async def github_webhook(request): """Handler for GitHub webhooks. Sends a Discord message when a new sponsor is created. Expects optional environment variables: - GITHUB_WEBHOOK_SECRET: webhook secret to verify signatures - GITHUB_WEBHOOK_CHANNEL_ID: Discord channel ID to post messages to """ try: body = await request.read() # Hardcoded secret (set here directly as requested) secret = "adasdjh8798asydhad" signature_header = request.headers.get("X-Hub-Signature-256", "") valid = await verify_github_signature(secret, body, signature_header) if not valid: logger.warning("GitHub webhook signature verification failed") return web.Response(status=403, text="Invalid signature") event = request.headers.get("X-GitHub-Event", "") payload = json.loads(body.decode("utf-8") or "{}") # Only handle sponsorship events for now if event == "sponsorship": action = payload.get("action") sponsorship = payload.get("sponsorship", {}) or {} sponsor = sponsorship.get("sponsor", {}) or {} sponsor_login = sponsor.get("login") or sponsor.get("email") or sponsor.get("name") sponsor_name = sponsor.get("name") or sponsor_login or "Someone" # Determine the channel id (hardcoded) chan_env = "1353495378368008264" try: channel_id = int(chan_env) except ValueError: logger.error("Invalid GITHUB_WEBHOOK_CHANNEL_ID value") return web.Response(status=500, text="Invalid channel id") # Helper to send message async def send_to_channel(msg: str): channel = bot.get_channel(channel_id) if channel is None: try: channel = await bot.fetch_channel(channel_id) except Exception: channel = None if not channel: logger.error(f"Discord channel {channel_id} not found") return False try: await channel.send(msg) return True except Exception as e: logger.error(f"Failed to send message to channel {channel_id}: {e}") return False # Handle actions if action == "created": message = f"**{sponsor_name}** just sponsored YASB on GitHub! Thank you! 🎉" ok = await send_to_channel(message) if ok: logger.info(f"Posted sponsorship message for {sponsor_login}") return web.Response(text="OK") else: return web.Response(status=500, text="Failed to send message") elif action == "edited": # Tier or details changed — notify optionally message = f"**{sponsor_name}** updated their sponsorship. Thank you! 🎉" await send_to_channel(message) return web.Response(text="OK") else: logger.info(f"Ignored sponsorship event action: {action}") return web.Response(text="Ignored") # For other event types, just acknowledge return web.Response(text="Event ignored") except Exception as e: logger.error(f"Error in GitHub webhook: {e}") return web.Response(status=500, text="Error") async def debug_catch_all(request): """Diagnostic handler that logs full request details for troubleshooting.""" try: # Log basic request-line info peername = request.transport.get_extra_info("peername") if request.transport else None logger.info(f"Incoming request from {peername}: {request.method} {request.path} {request.version}") # Log headers for k, v in request.headers.items(): logger.info(f"Header: {k}: {v}") # Read a limited amount of body for safety body = await request.read() snippet = body[:2048] try: logger.info(f"Body snippet (utf-8): {snippet.decode('utf-8', errors='replace')}") except Exception: logger.info(f"Body raw bytes: {snippet}") return web.Response(text="Debug logged") except Exception as e: logger.error(f"Error in debug_catch_all: {e}") return web.Response(status=500, text="Error") async def start_webserver(): app = web.Application() app.router.add_post("/kofi", kofi_webhook) app.router.add_post("/github", github_webhook) # Diagnostic catch-all for other requests (useful for debugging webhook deliveries) app.router.add_route("*", "/{tail:.*}", debug_catch_all) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, "0.0.0.0", 59856) await site.start() logger.info("Ko-fi webhook server started on port 59856") logger.info("GitHub webhook server started on port 59856") @bot.tree.command(name="test", description="Test if the YASBbot is working") async def ytest_slash(interaction: discord.Interaction): logger.info(f"/test command used by {interaction.user}") await interaction.response.send_message("YASBbot is working correctly!", ephemeral=True) @bot.tree.command(name="stats", description="Get stats and downloads info from the GitHub repo") async def stats_slash(interaction: discord.Interaction): logger.info(f"/stats command used by {interaction.user}") try: async with aiohttp.ClientSession() as session: repo_future = session.get("https://api.github.com/repos/amnweb/yasb") releases_future = session.get("https://api.github.com/repos/amnweb/yasb/releases") repo_response, releases_response = await asyncio.gather(repo_future, releases_future) if repo_response.status == 200 and releases_response.status == 200: repo_data = await repo_response.json() stars = repo_data.get("stargazers_count", 0) forks = repo_data.get("forks_count", 0) async with session.get("https://api.github.com/repos/amnweb/yasb/issues?state=open") as issues_resp: if issues_resp.status == 200: issues_data = await issues_resp.json() # Exclude pull requests issues = sum(1 for issue in issues_data if "pull_request" not in issue) else: issues = 0 releases_data = await releases_response.json() msi_downloads = 0 for release in releases_data: for asset in release.get("assets", []): if asset.get("name", "").endswith(".msi"): msi_downloads += asset.get("download_count", 0) funny_templates = [ "Buckle up! YASB's got **{stars}** stars lighting up the galaxy! Serving **{forks}** mouthwatering forks! Hunting down **{issues}** mischievous issues and **{msi_downloads:,}** downloads crashing our party! 🚀", "Party time! YASB boasts **{stars}** dazzling stars, dishes out **{forks}** epic forks, fumbles through **{issues}** quirky issues and scores **{msi_downloads:,}** downloads, let's celebrate! 🎉", "Look sharp! YASB shines with **{stars}** brilliant stars, dishes up **{forks}** scrumptious forks, zaps **{issues}** pesky issues and bags **{msi_downloads:,}** downloads in style! ✨", "Out of this world! YASB's rocking **{stars}** stars, tossing **{forks}** forks like meteors, battling **{issues}** alien issues and delivering **{msi_downloads:,}** downloads straight from the cosmos! 🛸", "Pop the bubbly! YASB flaunts **{stars}** sparkling stars, slices out **{forks}** irresistible forks, monkeying around with **{issues}** wild issues and handing out **{msi_downloads:,}** downloads, time to party! 🍾", ] message = random.choice(funny_templates).format( stars=stars, forks=forks, issues=issues, msi_downloads=msi_downloads ) await interaction.response.send_message(message, ephemeral=True) else: await interaction.response.send_message("Failed to fetch repo info and download stats!", ephemeral=True) except Exception as e: await interaction.response.send_message(f"Error fetching stats: {e}", ephemeral=True) @bot.event async def on_ready(): logger.info(f"Logged in as {bot.user.name}") logger.info("YASBbot is ready!") # Sync slash commands await bot.tree.sync() logger.info("Slash commands synced!") # Start the web server for Ko-fi webhook bot.loop.create_task(start_webserver()) @bot.event async def on_member_join(member): try: # Get the welcome channel welcome_channel = bot.get_channel(1355706362038059191) if welcome_channel: welcome_message = ( f"Hey {member.mention}, welcome to the party! 🎉\n\n" f"→ Quick peek at <#1353520965694390394>? It's like reading the manual, but fun!\n" f"→ Lost? Our [documentation ](https://docs.yasb.dev/latest) page is your treasure map!\n\n" f"Happy status bar tinkering! Breaking things is part of the experience 😄\n\n" ) await welcome_channel.send(welcome_message) logger.info(f"Sent welcome message for {member.name}") else: logger.error("Welcome channel not found!") except Exception as e: logger.error(f"Error in on_member_join: {e}") # Using token directly as requested bot.run("MTM1NTcyNzIwNjg0Nzg3MzEwNg.GkN6-X.6Pd4y5WB0YCFe9H-LfpI4NNuVpPYRvJGQWMch8")