FireBot/bot.py

235 lines
9 KiB
Python
Raw Normal View History

2023-11-05 01:37:35 +00:00
#!/usr/bin/python3
from socket import socket, AF_INET, SOCK_STREAM
from overrides import bytes, bbytes
2023-11-07 15:05:23 +00:00
import logs
2023-11-05 01:37:35 +00:00
import re
2023-11-07 13:25:53 +00:00
from typing import NoReturn, Union
2023-11-07 00:18:52 +00:00
import commands as cmds
2023-11-07 15:05:23 +00:00
import config as conf
2023-11-07 13:53:14 +00:00
from time import sleep
2023-11-07 15:05:23 +00:00
from importlib import reload
2023-11-08 00:30:06 +00:00
import random as r
import handlers
import bare
2023-11-06 00:50:04 +00:00
2023-11-08 02:30:35 +00:00
2023-11-07 13:25:53 +00:00
def mfind(message: str, find: list, usePrefix: bool = True) -> bool:
if usePrefix:
2023-11-07 15:05:23 +00:00
return any(message[: len(match) + 1] == conf.prefix + match for match in find)
2023-11-07 13:25:53 +00:00
else:
return any(message[: len(match)] == match for match in find)
2023-11-08 02:30:35 +00:00
class bot(bare.bot):
2023-11-06 22:23:31 +00:00
def __init__(self, server: str):
2023-11-05 01:37:35 +00:00
self.gmode = False
self.server = server
self.nicklen = 30
2023-11-07 15:05:23 +00:00
self.address = conf.servers[server]["address"]
2023-11-08 02:30:35 +00:00
self.port = (
conf.servers[server]["port"] if "port" in conf.servers[server] else 6667
)
2023-11-07 15:05:23 +00:00
self.channels = conf.servers[server]["channels"]
2023-11-05 01:37:35 +00:00
self.interval = (
2023-11-08 02:30:35 +00:00
conf.servers[server]["interval"]
if "interval" in conf.servers[server]
else 50
2023-11-05 01:37:35 +00:00
)
2023-11-07 15:05:23 +00:00
self.__version__ = conf.__version__
2023-11-06 03:08:47 +00:00
self.nick = "FireBot"
2023-11-07 15:05:23 +00:00
self.adminnames = conf.servers[server]["admins"]
2023-11-09 03:20:50 +00:00
self.queue: list[bbytes] = []
2023-11-06 02:36:33 +00:00
self.sock = socket(AF_INET, SOCK_STREAM)
self.npallowed = ["FireBitBot"]
2023-11-07 13:53:14 +00:00
self.log(f"Start init for {self.server}")
2023-11-05 01:37:35 +00:00
2023-11-06 00:49:57 +00:00
def connect(self) -> None:
2023-11-07 13:53:14 +00:00
self.log(f"Joining {self.server}...")
2023-11-06 02:36:33 +00:00
self.sock.connect((self.address, self.port))
2023-11-06 03:08:47 +00:00
self.send(f"USER {self.nick} {self.nick} {self.nick} {self.nick}\n")
self.send(f"NICK {self.nick}\n")
ircmsg = ""
while True:
ircmsg = self.recv().safe_decode()
2023-11-06 00:49:57 +00:00
if ircmsg != "":
2023-11-09 03:20:50 +00:00
code = 0
try:
code = int(ircmsg.split(" ", 2)[1].strip())
except (IndexError, ValueError):
2023-11-15 04:47:38 +00:00
pass
2023-11-06 00:49:57 +00:00
print(bytes(ircmsg).lazy_decode())
2023-11-15 00:50:26 +00:00
if "NICKLEN" in ircmsg:
2023-11-06 00:49:57 +00:00
self.nicklen = int(ircmsg.split("NICKLEN=")[1].split(" ")[0])
2023-11-07 13:53:14 +00:00
self.log(f"NICKLEN set to {self.nicklen}")
2023-11-15 00:37:46 +00:00
if code == 433:
2023-11-06 02:24:43 +00:00
self.log("Nickname in use", "WARN")
2023-11-06 03:08:47 +00:00
self.nick = f"{self.nick}{r.randint(0,1000)}"
self.send(f"NICK {self.nick}\n")
self.log(f"nick is now {self.nick}")
2023-11-15 00:37:46 +00:00
if code in [376, 422]:
self.log(f"Success by code: {code}")
break
2023-11-15 04:55:41 +00:00
if " MODE " in ircmsg or " PRIVMSG " in ircmsg:
self.log(f"Success by MSG/MODE")
break
2023-11-15 00:37:46 +00:00
if ircmsg.startswith("PING "):
2023-11-06 00:49:57 +00:00
self.ping(ircmsg)
2023-11-15 00:37:46 +00:00
if len(ircmsg.split("\x01")) == 3:
2023-11-09 21:36:34 +00:00
handlers.CTCP(self, ircmsg)
2023-11-15 00:50:26 +00:00
if "Closing link" in ircmsg:
2023-11-06 00:49:57 +00:00
self.exit("Closing Link")
else:
self.exit("Lost connection to the server")
2023-11-07 13:53:14 +00:00
self.log(f"Joined {self.server} successfully!")
2023-11-06 00:49:57 +00:00
2023-11-07 00:18:52 +00:00
def join(self, chan: str, origin: str, lock: bool = True) -> None:
2023-11-06 22:20:54 +00:00
self.log(f"Joining {chan}...")
2023-11-06 03:26:41 +00:00
chan = chan.replace(" ", "")
if "," in chan:
chans = chan.split(",")
for subchan in chans:
2023-11-06 22:20:54 +00:00
self.join(subchan, origin)
2023-11-06 03:26:41 +00:00
return
if chan.startswith("0") or (chan == "#main" and lock):
if origin != "null":
2023-11-08 04:23:40 +00:00
self.msg(f"Refusing to join channel {chan} (protected)", origin)
2023-11-06 03:26:41 +00:00
return
2023-11-07 13:53:14 +00:00
if chan in self.channels and lock:
2023-11-06 03:26:41 +00:00
if origin != "null":
2023-11-08 04:23:40 +00:00
self.msg(f"I'm already in {chan}.", origin)
2023-11-06 03:26:41 +00:00
return
2023-11-06 22:20:54 +00:00
self.send(f"JOIN {chan}\n")
2023-11-06 03:26:41 +00:00
while True:
ircmsg = self.recv().safe_decode()
2023-11-06 03:26:41 +00:00
if ircmsg != "":
code = 0
try:
code = int(ircmsg.split(" ", 2)[1].strip())
except (IndexError, ValueError):
pass
2023-11-06 03:26:41 +00:00
print(bytes(ircmsg).lazy_decode())
if ircmsg.startswith("PING "):
self.ping(ircmsg)
2023-11-06 03:26:41 +00:00
elif len(ircmsg.split("\x01")) == 3:
2023-11-09 21:36:34 +00:00
handlers.CTCP(self, ircmsg)
elif code == 403:
2023-11-06 03:26:41 +00:00
self.log(f"Joining {chan} failed", "WARN")
if origin != "null":
self.msg(f"{chan} is an invalid channel", origin)
2023-11-06 03:26:41 +00:00
break
elif code == 473:
self.log(f"Joining {chan} failed (+i)", "WARN")
2023-11-06 03:26:41 +00:00
if origin != "null":
self.msg(f"{chan} is +i, and I'm not invited.", origin)
2023-11-06 03:26:41 +00:00
break
2023-11-06 22:04:46 +00:00
elif code == 520:
self.log(f"Joining {chan} failed (+O)", "WARN")
if origin != "null":
self.msg(f"{chan} is +O, and I'm not an operator.", origin)
break
elif code == 405:
self.log(f"Joining {chan} failed (too many channels)", "WARN")
if origin != "null":
self.msg(f"I'm in too many channels to join {chan}", origin)
break
elif code == 471:
self.log(f"Joining {chan} failed (+l)", "WARN")
if origin != "null":
self.msg(f"{chan} is +l, and is full", origin)
break
elif code == 366:
2023-11-07 13:53:14 +00:00
self.log(f"Joining {chan} succeeded")
2023-11-06 03:26:41 +00:00
if origin != "null":
self.msg(f"Joined {chan}", origin)
2023-11-06 03:26:41 +00:00
self.channels[chan] = 0
break
2023-11-07 13:59:01 +00:00
def ping(self, ircmsg: str) -> int:
pong = f"PONG :{ircmsg.split('PING :')[1]}\n"
print(pong, end="")
return self.send(pong)
2023-11-05 01:37:35 +00:00
def send(self, command: str) -> int:
2023-11-06 22:20:54 +00:00
return self.sock.send(bytes(command))
2023-11-05 01:37:35 +00:00
def recv(self) -> bytes:
2023-11-05 22:26:50 +00:00
if self.queue:
return bytes(self.queue.pop(0))
2023-11-06 02:36:33 +00:00
data = bytes(self.sock.recv(2048).strip(b"\r\n"))
2023-11-05 22:26:50 +00:00
if b"\r\n" in data:
self.queue.extend(data.split(b"\r\n"))
return bytes(self.queue.pop(0))
return data
2023-11-05 01:37:35 +00:00
2023-11-09 03:20:50 +00:00
def log(self, message: str, level: str = "LOG") -> None:
2023-11-07 15:05:23 +00:00
logs.log(message, self.server)
2023-11-05 01:37:35 +00:00
2023-11-09 03:20:50 +00:00
def exit(self, message: str) -> NoReturn:
2023-11-07 15:05:23 +00:00
logs.log(message, self.server, "EXIT")
2023-11-05 01:37:35 +00:00
exit(1)
2023-11-06 06:29:00 +00:00
2023-11-07 13:25:53 +00:00
def msg(self, msg: str, target: str) -> None:
if not (target == "NickServ" and mfind(msg, ["IDENTIFY"], False)):
self.log(f"Sending {bytes(msg).lazy_decode()} to {target}")
else:
self.log("Identifying myself...")
self.send(f"PRIVMSG {target} :{msg}\n")
def op(self, name: str, chan: str) -> Union[int, None]:
if name != "":
self.log(f"Attempting op of {name} in {chan}...")
return self.send(f"MODE {chan} +o {name}\n")
2023-11-09 03:20:50 +00:00
return None
2023-11-07 13:25:53 +00:00
def notice(self, msg: str, target: str, silent: bool = False) -> int:
if not silent:
self.log(f"Sending {bytes(msg).lazy_decode()} to {target} (NOTICE)")
return self.send(f"NOTICE {target} :{msg}\n")
def sendraw(self, command: str) -> int:
self.log(f"RAW sending {command}")
command = f"{command}\n"
return self.send(command.replace("$BOTNICK", self.nick))
2023-11-07 13:25:53 +00:00
2023-11-07 00:18:52 +00:00
def mainloop(self) -> NoReturn:
self.log("Starting connection..")
self.connect()
2023-11-07 15:05:23 +00:00
if "pass" in conf.servers[self.server]:
2023-11-08 02:30:35 +00:00
self.msg(
f"IDENTIFY FireBot {conf.servers[self.server]['pass']}", "NickServ"
)
2023-11-07 00:18:52 +00:00
sleep(0.5)
for chan in self.channels:
self.join(chan, "null", False)
while 1:
raw = self.recv()
ircmsg = raw.safe_decode()
2023-11-07 00:18:52 +00:00
if ircmsg == "":
2023-11-07 13:59:01 +00:00
self.exit("Probably a netsplit")
2023-11-07 00:18:52 +00:00
else:
print(raw.lazy_decode(), sep="\n")
action = "Unknown"
try:
action = ircmsg.split(" ", 2)[1].strip()
except IndexError:
pass
if action == "PRIVMSG":
2023-11-09 21:42:17 +00:00
res, chan = handlers.PRIVMSG(self, ircmsg)
if res == "reload":
reload(conf)
reload(cmds)
reload(handlers)
2023-11-13 21:42:46 +00:00
self.msg("Reloaded successfully", chan) # type: ignore
2023-11-08 02:26:06 +00:00
elif action == "NICK":
name = ircmsg.split("!", 1)[0][1:]
if name == self.nick:
self.nick = ircmsg.split("NICK", 1)[1].split(":", 1)[1].strip()
2023-11-07 14:01:37 +00:00
else:
if ircmsg.startswith("PING "):
2023-11-07 14:02:05 +00:00
self.ping(ircmsg)
2023-11-07 14:01:37 +00:00
elif ircmsg.startswith("ERROR :Closing Link"):
2023-11-07 14:02:05 +00:00
self.exit("I got killed :'(")
2023-11-07 14:01:37 +00:00
elif ircmsg.startswith("ERROR :Ping "):
2023-11-07 14:02:05 +00:00
self.exit("Ping timeout")
2023-11-09 03:24:43 +00:00
self.exit("While loop broken")