Skip to content

Commit

Permalink
updated cogs to use config and fixed access to restricted attribute u…
Browse files Browse the repository at this point in the history 
…ses of bot._http
  • Loading branch information
@minimike86
minimike86 committed Nov 3, 2024
1 parent 32ca83c commit 743ce09
Show file tree
Hide file tree
Showing 9 changed files with 274 additions and 239 deletions.
5 changes: 3 additions & 2 deletions twitchrce/api/twitch/twitch_api_auth.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -174,11 +174,12 @@ async def validate_token(access_token: str) -> bool:
return True

async def get_users(
self, access_token: str, ids=Optional[int], logins=Optional[str]
self, access_token: str, _ids=Optional[int], _logins=Optional[str]
):
"""
https://dev.twitch.tv/docs/authentication/validate-tokens/
WARNING Twitch periodically conducts audits to discover applications that are not validating access tokens hourly as required.
WARNING Twitch periodically conducts audits to discover applications that are not validating access tokens
hourly as required.
"""
url = "https://api.twitch.tv/helix/users"
headers = {
Expand Down
144 changes: 72 additions & 72 deletions twitchrce/cogs/rce.py
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import logging

Check failure on line 1 in twitchrce/cogs/rce.py

View workflow job for this annotation

GitHub Actions / Build 

Imports are incorrectly sorted and/or formatted.
import re
import shlex
import subprocess
Expand All @@ -8,53 +9,58 @@

import twitchio
from colorama import Fore, Style
from twitchio import errors
from custom_bot import CustomBot
from twitchio import User, errors, PartialUser, PartialChatter, Chatter
from twitchio.ext import commands, pubsub

from twitchrce import custom_bot
from twitchrce.config import bot_config
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s | %(levelname)-8s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)


class RCECog(commands.Cog):

def __init__(self, bot: custom_bot.CustomBot):
def __init__(self, bot: CustomBot):
self.bot = bot

@commands.Cog.event()
async def event_message(self, message: twitchio.Message):
if message.echo:
return
# print('RCECog: ', message.author.name, message.content)
logger.info("RCECog: ", message.author.name, message.content)

@commands.command(aliases=["cmd"])
async def exec(self, ctx: commands.Context):
# get channel broadcaster
broadcaster = await self.bot._http.get_users(ids=[], logins=[ctx.channel.name])
user_access_token_resultset = self.bot.database.fetch_user_access_token(
broadcaster_id=self.bot.user_id
)
broadcaster: User = (
await self.bot.fetch_users(ids=[], logins=[ctx.channel.name])
)[0]

if ctx.message.content == "!exec --help":
await ctx.send(
"""exec: !exec [whatever /bin/bash commands you want to mess with the streamer]:
This will run (mostly) un-sanitised bash commands on the streamers machine. rm -rf for the win."""
This will run (mostly) un-sanitised bash commands on the streamers machine. rm -rf for the win."""
)

# only broadcaster can run exec commands
# TODO: allow mods to run exec commands
elif (
int(ctx.author.id) == int(broadcaster[0]["id"])
or int(ctx.author.id) == 125444292
):
# only broadcaster or bot can run exec commands
elif ctx.author.id == broadcaster.id or ctx.author.id == self.bot.bot_user.id:
# grab the arbitrary bash command(s) without the bot prefix
cmd = None
if ctx.message.content[:5] == "!exec":
cmd = re.sub(
rf"^{self.bot._prefix}{ctx.command.name}", "", ctx.message.content
rf"^{self.bot.get_prefix(message=ctx.message)}{ctx.command.name}",
"",
ctx.message.content,
).strip()
else:
for alias in ctx.command.aliases:
cmd = re.sub(
rf"^{self.bot._prefix}{alias}", "", ctx.message.content
rf"^{self.bot.get_prefix(message=ctx.message)}{alias}",
"",
ctx.message.content,
).strip()

# strip operators
Expand All @@ -73,7 +79,10 @@ async def exec(self, ctx: commands.Context):
command1 = shlex.split(cmd1)
command2 = shlex.split(cmd2)
pass
if command1[0] in settings.CMD_ALLOW_LIST and command2[0] == "grep":
if (
command1[0] in self.bot.config.CMD_ALLOW_LIST
and command2[0] == "grep"
):
proc1 = Popen(
command1[:4],
shell=False,
Expand All @@ -93,7 +102,7 @@ async def exec(self, ctx: commands.Context):
"""if input has no pipes run the command"""
command = shlex.split(cmd)
pass
if command[0] in settings.CMD_ALLOW_LIST:
if command[0] in self.bot.config.CMD_ALLOW_LIST:
proc = Popen(
command, shell=False, stdin=PIPE, stdout=PIPE, stderr=PIPE
)
Expand All @@ -106,10 +115,9 @@ async def exec(self, ctx: commands.Context):
"""post the stdout to chat"""
stdout = f"stdout: {stdout.decode()}"
try:
await self.bot._http.post_chat_announcement(
token=user_access_token_resultset["access_token"],
broadcaster_id=str(broadcaster[0]["id"]),
moderator_id=self.bot.user_id,
await self.bot.post_chat_announcement(
broadcaster=broadcaster,
moderator=self.bot.bot_user,
message=f"{textwrap.shorten(stdout, width=500)}",
color="green",
)
Expand All @@ -119,10 +127,9 @@ async def exec(self, ctx: commands.Context):
"""post the stderr to chat"""
stderr = f"stderr: {stderr.decode()}"
try:
await self.bot._http.post_chat_announcement(
token=user_access_token_resultset["access_token"],
broadcaster_id=str(broadcaster[0]["id"]),
moderator_id=self.bot.user_id,
await self.bot.post_chat_announcement(
broadcaster=broadcaster,
moderator=self.bot.bot_user,
message=f"{textwrap.shorten(stderr, width=500)}",
color="orange",
)
Expand All @@ -131,12 +138,15 @@ async def exec(self, ctx: commands.Context):

else:
"""post message to chat informing they tried to run a command that wasn't in the allow list"""
error_msg = f"Nice try {ctx.author.display_name} but the command(s) in `{cmd}` are not in the allow list!"
# noinspection DuplicatedCode
error_msg = (
f"Nice try {ctx.author.display_name} but the command(s) in '{cmd}' "
f"are not in the allow list!"
)
try:
await self.bot._http.post_chat_announcement(
token=user_access_token_resultset["access_token"],
broadcaster_id=str(broadcaster[0]["id"]),
moderator_id=self.bot.user_id,
await self.bot.post_chat_announcement(
broadcaster=broadcaster,
moderator=self.bot.bot_user,
message=f"{textwrap.shorten(error_msg, width=500)}",
color="orange",
)
Expand All @@ -147,12 +157,15 @@ async def exec(self, ctx: commands.Context):
await ctx.channel.send("TimeoutError occurred")
except subprocess.TimeoutExpired:
"""post message to chat informing they tried to run a command that took too long to run"""
error_msg = f"Nice try {ctx.author.display_name} but the command(s) in `{cmd}` took too long to run!"
# noinspection DuplicatedCode
error_msg = (
f"Nice try {ctx.author.display_name} but the command(s) in '{cmd}' "
f"took too long to run!"
)
try:
await self.bot._http.post_chat_announcement(
token=user_access_token_resultset["access_token"],
broadcaster_id=str(broadcaster[0]["id"]),
moderator_id=self.bot.user_id,
await self.bot.post_chat_announcement(
broadcaster=broadcaster,
moderator=self.bot.bot_user,
message=f"{textwrap.shorten(error_msg, width=500)}",
color="orange",
)
Expand All @@ -167,25 +180,12 @@ async def exec(self, ctx: commands.Context):
except RuntimeError:
await ctx.channel.send("An exception occurred")

async def killmyshell(
async def kill_my_shell(
self,
broadcaster_id: int,
author_login: str,
broadcaster: User | PartialUser,
chatter: Chatter | PartialChatter,
event: pubsub.PubSubChannelPointsMessage,
):
# get channel broadcaster
broadcaster = await self.bot._http.get_users(
ids=[str(broadcaster_id)], logins=[]
)
broadcaster_access_token_resultset = self.bot.database.fetch_user_access_token(
broadcaster_id=broadcaster[0]["id"]
)
broadcaster_access_token = broadcaster_access_token_resultset["access_token"]
mod_access_token_resultset = self.bot.database.fetch_user_access_token(
broadcaster_id=self.bot.user_id
)
mod_access_token = mod_access_token_resultset["access_token"]

cmd1 = "echo $(xwininfo -tree -root | grep qterminal | head -n 1)"
proc_id = (
subprocess.check_output(cmd1, shell=True).decode().split(" ")[0].strip()
Expand All @@ -195,22 +195,20 @@ async def killmyshell(
cmd2 = f"xkill -id {proc_id}"
result = subprocess.check_output(cmd2, shell=True)
result = (
f"{Fore.RED}{author_login} just killed {broadcaster[0]['display_name']}'s shell. "
f"{Fore.RED}{chatter.display_name} just killed {broadcaster.name}'s shell. "
+ f"{Fore.MAGENTA}stdout: {result.decode()}{Style.RESET_ALL}"
)
try:
await self.bot._http.update_reward_redemption_status(
token=broadcaster_access_token,
broadcaster_id=str(broadcaster[0]["id"]),
await self.bot.update_reward_redemption_status(
broadcaster=broadcaster,
reward_id=event.id,
custom_reward_id=event.reward.id,
status=True,
)
print(f"{textwrap.shorten(result, width=500)}")
await self.bot._http.post_chat_announcement(
token=mod_access_token,
broadcaster_id=str(broadcaster[0]["id"]),
moderator_id=self.bot.user_id,
await self.bot.post_chat_announcement(
broadcaster=broadcaster,
moderator=self.bot.bot_user,
message=f"{textwrap.shorten(result, width=500)}",
color="green",
)
Expand All @@ -224,24 +222,26 @@ async def killmyshell(
)
else:
try:
await self.bot._http.update_reward_redemption_status(
token=broadcaster_access_token,
broadcaster_id=str(broadcaster[0]["id"]),
await self.bot.update_reward_redemption_status(
broadcaster=broadcaster,
reward_id=event.id,
custom_reward_id=event.reward.id,
status=False,
)
print(
f"{Fore.RED}Unlucky {author_login} but there are no terminals open to kill{Style.RESET_ALL}"
f"{Fore.RED}Unlucky {chatter.display_name} but there are no terminals open to kill{Style.RESET_ALL}"
)
message = (
f"Unlucky {chatter.display_name} there's no terminals open to kill so your channel points have "
f"been refunded"
)
await self.bot._http.post_chat_announcement(
token=mod_access_token,
broadcaster_id=str(broadcaster[0]["id"]),
moderator_id=self.bot.user_id,
message=f"Unlucky {author_login} there are no terminals open to kill; your channel points have been refunded",
await self.bot.post_chat_announcement(
broadcaster=broadcaster,
moderator=self.bot.bot_user,
message=message,
color="orange",
)
except errors.AuthenticationError:
print(
f"{Fore.RED}Unlucky {author_login} but there are no terminals open to kill{Style.RESET_ALL}"
f"{Fore.RED}Unlucky {chatter.display_name} but there are no terminals open to kill{Style.RESET_ALL}"
)
14 changes: 7 additions & 7 deletions twitchrce/cogs/user_cog.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,48 +21,48 @@ async def event_message(self, message: twitchio.Message):
async def stairsthetrashman1(self, ctx: commands.Context):
"""type !stairsthetrashman or !ohlook"""
if ctx.author.display_name.lower() in ["stairsthetrashman", "msec"]:
sound_file = "/home/kali/Music/ohlook.mp3"
_sound_file = "/home/kali/Music/ohlook.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)

@commands.command(aliases=["stairs2", "because"])
async def stairsthetrashman2(self, ctx: commands.Context):
if ctx.author.display_name.lower() in ["stairsthetrashman", "msec"]:
sound_file = "/home/kali/Music/Because_Im_mexican.mp3"
_sound_file = "/home/kali/Music/Because_Im_mexican.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)

@commands.command(aliases=["stairs3", "sonofagun"])
async def stairsthetrashman3(self, ctx: commands.Context):
if ctx.author.display_name.lower() in ["stairsthetrashman", "msec"]:
sound_file = "/home/kali/Music/yousonofagun.mp3"
_sound_file = "/home/kali/Music/yousonofagun.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)

@commands.command(aliases=["alh4zr3d", "alhashes"])
async def alh4zr3d1(self, ctx: commands.Context):
if ctx.author.display_name.lower() in ["alh4zr3d", "msec"]:
sound_file = "/home/kali/Music/al-fully-erect.mp3"
_sound_file = "/home/kali/Music/al-fully-erect.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)

@commands.command(aliases=["tibs", "0xtib3rius", "sofgood"])
async def tibs1(self, ctx: commands.Context):
if ctx.author.display_name.lower() in ["0xtib3rius", "msec"]:
sound_file = "/home/kali/Music/tibs1_sofgood.mp3"
_sound_file = "/home/kali/Music/tibs1_sofgood.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)

@commands.command(aliases=["trshpuppy", "sbacunt"])
async def trshpuppy1(self, ctx: commands.Context):
if ctx.author.display_name.lower() in ["trshpuppy", "msec"]:
sound_file = "/home/kali/Music/trashpuppy_stop-being-a-cunt.mp3"
_sound_file = "/home/kali/Music/trashpuppy_stop-being-a-cunt.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)

@commands.command(aliases=["fcamerayo"])
async def trshpuppy2(self, ctx: commands.Context):
if ctx.author.display_name.lower() in ["trshpuppy", "msec"]:
sound_file = "/home/kali/Music/trashpuppy_fucking-camera-yo.mp3"
_sound_file = "/home/kali/Music/trashpuppy_fucking-camera-yo.mp3"
# data, samplerate = sf.read(sound_file)
# self.bot.sd.play(data, samplerate)
Loading

0 comments on commit 743ce09

Please sign in to comment.