forked from Firepup650/FireBot
Error handling and dnsbl improvements
This commit is contained in:
parent
58060a5249
commit
8b41a048c5
5 changed files with 43 additions and 19 deletions
7
bot.py
7
bot.py
|
@ -14,6 +14,7 @@ import handlers
|
|||
import bare
|
||||
from threading import Thread
|
||||
from markov import MarkovBot
|
||||
from traceback import format_exc
|
||||
|
||||
|
||||
def mfind(message: str, find: list, usePrefix: bool = True) -> bool:
|
||||
|
@ -317,6 +318,7 @@ class bot(bare.bot):
|
|||
if action in handlers.handles:
|
||||
res, chan = handlers.handles[action](self, ircmsg)
|
||||
if res == "reload" and type(chan) == str:
|
||||
try:
|
||||
reload(conf)
|
||||
self.adminnames = (
|
||||
conf.servers[self.server]["admins"]
|
||||
|
@ -343,6 +345,11 @@ class bot(bare.bot):
|
|||
reload(cmds)
|
||||
reload(handlers)
|
||||
self.msg("Reloaded successfully", chan)
|
||||
except Exception:
|
||||
Err = format_exc()
|
||||
for line in Err.split("\n"):
|
||||
self.log(line, "ERROR")
|
||||
self.msg("Reload failed, likely partially reloaded. Please check error logs.", chan)
|
||||
else:
|
||||
if ircmsg.startswith("PING "):
|
||||
self.ping(ircmsg)
|
||||
|
|
12
commands.py
12
commands.py
|
@ -281,8 +281,8 @@ def check(bot: bare.bot, chan: str, name: str, message: str) -> None:
|
|||
msg = message.split(' ', 1)[1]
|
||||
nick = msg.split('!')[0]
|
||||
host = msg.split('@', 1)[1]
|
||||
dnsbl = conf.dnsblHandler(bot, nick, host, chan)
|
||||
bot.msg("Blacklist check: " + (dnsbl if dnsbl else "Safe."), chan)
|
||||
dnsbl, raws = conf.dnsblHandler(bot, nick, host, chan)
|
||||
bot.msg(f"Blacklist check: {dnsbl if dnsbl else 'Safe.'} ({raws})", chan)
|
||||
except Exception as E:
|
||||
bot.msg("Blacklist lookup failed. Error recorded to bot logs.", chan)
|
||||
bot.log(str(E), "FATAL")
|
||||
|
@ -299,12 +299,12 @@ data: dict[str, dict[str, Any]] = {
|
|||
"check": checks.admin,
|
||||
},
|
||||
"uptime": {"prefix": True, "aliases": []},
|
||||
"raw ": {"prefix": True, "aliases": ["cmd "], "check": checks.admin},
|
||||
"raw": {"prefix": True, "aliases": ["cmd "], "check": checks.admin},
|
||||
"debug": {"prefix": True, "aliases": ["dbg", "d"], "check": checks.admin},
|
||||
"debugInternal": {"prefix": True, "aliases": ["dbgInt", "dI"], "check": checks.admin},
|
||||
"debugEval": {"prefix": True, "aliases": ["dbgEval", "dE"], "check": checks.admin},
|
||||
"8ball": {"prefix": True, "aliases": ["eightball", "8b"]},
|
||||
"join ": {"prefix": True, "aliases": [], "check": checks.admin},
|
||||
"join": {"prefix": True, "aliases": ["j"], "check": checks.admin},
|
||||
"quote": {"prefix": True, "aliases": ["q"]},
|
||||
"goat.mode.activate": {"prefix": True, "aliases": ["g.m.a"], "check": checks.admin},
|
||||
"goat.mode.deactivate": {
|
||||
|
@ -334,12 +334,12 @@ call: dict[str, Callable[[bare.bot, str, str, str], None]] = {
|
|||
conf.su: sudo,
|
||||
"restart": reboot,
|
||||
"uptime": uptime,
|
||||
"raw ": raw,
|
||||
"raw": raw,
|
||||
"debug": debug,
|
||||
"debugInternal": debugInternal,
|
||||
"debugEval": debugEval,
|
||||
"8ball": eball,
|
||||
"join ": join,
|
||||
"join": join,
|
||||
"quote": quote,
|
||||
"goat.mode.activate": goatOn,
|
||||
"goat.mode.decativate": goatOff,
|
||||
|
|
24
config.py
24
config.py
|
@ -4,10 +4,12 @@ from dotenv import load_dotenv # type: ignore
|
|||
import re, codecs
|
||||
from typing import Optional, Any, Union
|
||||
import bare, pylast
|
||||
from pydnsbl import DNSBLIpChecker, DNSBLDomainChecker
|
||||
from pydnsbl import DNSBLIpChecker, DNSBLDomainChecker, providers as BL
|
||||
|
||||
ipbl = DNSBLIpChecker()
|
||||
hsbl = DNSBLDomainChecker()
|
||||
providers = BL.BASE_PROVIDERS + [BL.Provider('dnsbl.dronebl.org.org')]
|
||||
|
||||
ipbl = DNSBLIpChecker(providers=providers)
|
||||
hsbl = DNSBLDomainChecker(providers=providers)
|
||||
|
||||
load_dotenv()
|
||||
__version__ = "v3.0.16"
|
||||
|
@ -46,6 +48,7 @@ servers: dict[str, dict[str, Any]] = {
|
|||
"#fp-radio": 0,
|
||||
"#fp-radio-debug": 0,
|
||||
"#hardfork": 0,
|
||||
"#opers": 0,
|
||||
},
|
||||
"ignores": ["#fp-radio"],
|
||||
"admins": ["h-tl"],
|
||||
|
@ -137,9 +140,9 @@ def sub(
|
|||
return result
|
||||
|
||||
|
||||
def dnsbl(hostname: str) -> str:
|
||||
def dnsbl(hostname: str) -> tuple[str, dict[str, list[str]]]:
|
||||
hosts = []
|
||||
hstDT = None
|
||||
hstDT = {}
|
||||
try:
|
||||
hstDT = ipbl.check(hostname).detected_by
|
||||
except ValueError: # It's not an IP
|
||||
|
@ -151,18 +154,19 @@ def dnsbl(hostname: str) -> str:
|
|||
if hstDT[host] != ["unknown"]:
|
||||
hosts.append(host)
|
||||
if not hosts:
|
||||
return ""
|
||||
return "", hstDT
|
||||
hostStr = None
|
||||
if len(hosts) >= 3:
|
||||
hostStr = ', and '.join((', '.join(hosts)).rsplit(", ", 1))
|
||||
else:
|
||||
hostStr = ' and '.join(hosts)
|
||||
return hostStr
|
||||
return hostStr, hstDT
|
||||
|
||||
def dnsblHandler(bot: bare.bot, nick: str, hostname: str, chan: str) -> str:
|
||||
def dnsblHandler(bot: bare.bot, nick: str, hostname: str, chan: str) -> tuple[str, dict[str, list[str]]]:
|
||||
dnsblStatus = 'Not enabled'
|
||||
dnsblResps = {}
|
||||
if bot.dnsblMode != "none":
|
||||
dnsblStatus = dnsbl(hostname)
|
||||
dnsblStatus, dnsblResps = dnsbl(hostname)
|
||||
if dnsblStatus:
|
||||
match bot.dnsblMode:
|
||||
case "kickban":
|
||||
|
@ -180,4 +184,4 @@ def dnsblHandler(bot: bare.bot, nick: str, hostname: str, chan: str) -> str:
|
|||
bot.sendraw(f"GLINE *@{hostname} :Sorry, but you're on the {dnsblStatus} blacklist(s).")
|
||||
case _:
|
||||
bot.log(f'Unknown dnsbl Mode "{bot.dnsblMode}"!', "WARN")
|
||||
return dnsblStatus
|
||||
return dnsblStatus, dnsblResps
|
||||
|
|
17
handlers.py
17
handlers.py
|
@ -6,6 +6,7 @@ from typing import Union, Callable
|
|||
from overrides import bytes, bbytes
|
||||
from importlib import reload
|
||||
import bare, re, checks
|
||||
from traceback import format_exc
|
||||
|
||||
|
||||
def CTCP(bot: bare.bot, msg: str) -> bool:
|
||||
|
@ -110,9 +111,21 @@ def PRIVMSG(bot: bare.bot, msg: str) -> Union[tuple[None, None], tuple[str, str]
|
|||
):
|
||||
if "check" in cmds.data[cmd] and cmds.data[cmd]["check"]:
|
||||
if cmds.data[cmd]["check"](bot, name, host, chan, cmd):
|
||||
cmds.call[cmd](bot, chan, name, message)
|
||||
try:
|
||||
cmds.call[cmd](bot, chan, name, message)
|
||||
except Exception:
|
||||
Err = format_exc()
|
||||
for line in Err.split("\n"):
|
||||
bot.log(line, "ERROR")
|
||||
bot.msg("Sorry, I had an error trying to execute that command. Please check error logs.", chan)
|
||||
else:
|
||||
cmds.call[cmd](bot, chan, name, message)
|
||||
try:
|
||||
cmds.call[cmd](bot, chan, name, message)
|
||||
except Exception:
|
||||
Err = format_exc()
|
||||
for line in Err.split("\n"):
|
||||
bot.log(line, "ERROR")
|
||||
bot.msg("Sorry, I had an error trying to execute that command. Please check error logs.", chan)
|
||||
handled = True
|
||||
break
|
||||
if not handled:
|
||||
|
|
2
logs.py
2
logs.py
|
@ -10,7 +10,7 @@ def log(
|
|||
level: str = "LOG",
|
||||
time: Union[dt, str] = "now",
|
||||
) -> None:
|
||||
if level in ["EXIT", "CRASH", "FATAL"]:
|
||||
if level in ["EXIT", "CRASH", "FATAL", "ERROR"]:
|
||||
stream = stderr
|
||||
else:
|
||||
stream = stdout
|
||||
|
|
Loading…
Reference in a new issue