Pathogen
/
monolith
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Improve authentication detection 

Add a negative check in the event we are authenticated and registered,
but not confirmed, as this fools other checks.
This commit is contained in:
Mark Veidemanis 2020-10-31 16:49:37 +00:00
parent eaeb4b72c2
commit d60d89dbf6
6 changed files with 125 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
commands/authcheck.py
View File

@ -16,8 +16,8 @@ class AuthcheckCommand:
info("\n".join(results))
return
elif length == 2:
if not spl[1] in main.IRCPool.keys():
failure("No such instance: %s" % spl[1])
if not spl[1] in main.network.keys():
failure("No such network: %s" % spl[1])
return
results = []
for i in main.IRCPool.keys():

36
commands/recheckauth.py Normal file
View File

@ -0,0 +1,36 @@
import main
class RecheckauthCommand:
def __init__(self, *args):
self.recheckauth(*args)
def recheckauth(self, addr, authed, data, obj, spl, success, failure, info, incUsage, length):
if authed:
if length == 1:
results = []
for i in main.IRCPool.keys():
num = main.IRCPool[i].num
net = main.IRCPool[i].net
main.IRCPool[i].authenticated = False
main.IRCPool[i].regPing()
success("Successfully reset authentication status")
return
elif length == 2:
if not spl[1] in main.network.keys():
failure("No such network: %s" % spl[1])
return
results = []
for i in main.IRCPool.keys():
num = main.IRCPool[i].num
net = main.IRCPool[i].net
if not net == spl[1]:
continue
main.IRCPool[i].authenticated = False
main.IRCPool[i].regPing()
success("Successfully reset authentication status on %s" % spl[1])
return
else:
incUsage("recheckauth")
return
else:
incUsage(None)

13
conf/example/irc.json
View File

@ -3,14 +3,21 @@
"register": true,
"entity": "NickServ",
"email": "{nickname}@domain.com",
"register": "REGISTER {password} {email}",
"registermsg": "REGISTER {password} {email}",
"confirm": "CONFIRM {token}",
"check": true,
"ping": true,
"pingmsg": "STATUS {nickname}",
"negative": true,
"pingmsg": "STATUS",
"negativemsg": "INFO {nickname}",
"checktype": "mode",
"checkmode": "r",
"checkmsg": "Password accepted - you are now recognized.",
"checkmsg2": "You are logged in as"
"checkmsg2": "You are logged in as",
"checknegativemsg": "has \u0002NOT COMPLETED\u0002 registration verification",
"checkendnegative": "End of Info"
},
"freenode": {
"confirm": "VERIFY REGISTER {nickname} {token}"
}
}

3
conf/help.json
View File

@ -31,5 +31,6 @@
"reg": "reg <network> <num>",
"confirm": "confirm <network> <num> <token>",
"pending": "pending [<network>]",
"authcheck": "authcheck [<network>]"
"authcheck": "authcheck [<network>]",
"recheckauth": "recheckauth [<network>]"
}

40
core/bot.py
View File

@ -160,6 +160,7 @@ class IRCBot(IRCClient):
self.servername = None
self._regAttempt = None
self._negativePass = None
def lineReceived(self, line):
if bytes != str and isinstance(line, bytes):
@ -167,7 +168,6 @@ class IRCBot(IRCClient):
line = line.decode("utf-8", "replace")
line = lowDequote(line)
trace(self.net, self.num, line)
try:
prefix, command, params = parsemsg(line)
if command in numeric_to_symbolic:
@ -618,14 +618,46 @@ class IRCBot(IRCClient):
self.topicUpdated(prefix, channel, newtopic)
# End of Twisted hackery
def regPing(self, negativepass=None):
if self.authenticated:
return
sinst = regproc.substitute(self.net, self.num)
if not self._negativePass == True:
if negativepass == False:
self._negativePass = False
return
if negativepass == True:
if self._negativePass == None:
self._negativePass = True
debug("Positive registration check - %s - %i" % (self.net, self.num))
if sinst["ping"]:
debug("Sending ping - %s - %i" % (self.net, self.num))
self.msg(sinst["entity"], sinst["pingmsg"])
return
else:
debug("Negative registration for %s - %i" % (self.net, self.num))
return
if sinst["check"]:
if sinst["negative"]:
self._negativePass = None
self.msg(sinst["entity"], sinst["negativemsg"])
return
else:
self._negativePass = True
if sinst["ping"]:
self.msg(sinst["entity"], sinst["pingmsg"])
return
else:
self.authenticated = True
warn("Won't send registration ping for %s - %i" % (self.net, self.num))
def signedOn(self):
log("signed on: %s - %i" % (self.net, self.num))
ctime = str(datetime.now().isoformat())
sendRelayNotification({"type": "conn", "net": self.net, "num": self.num, "status": "signedon", "time": ctime})
if not self.authenticated:
inst = regproc.selectInst(self.net)
if inst["ping"] and inst["check"]:
self.msg(inst["entity"], inst["pingmsg"])
reactor.callLater(10, self.regPing)
def joined(self, channel):
if not channel in self.channels:

60
modules/regproc.py
View File

@ -22,38 +22,45 @@ def selectInst(net):
inst = main.irc["_"]
return inst
def registerAccount(net, num):
debug("Attempting to register: %s - %i" % (net, num))
def substitute(net, num, token=None):
inst = selectInst(net)
alias = main.alias[num]
nickname = alias["nick"]
username = nickname+"/"+net
password = main.network[net].aliases[num]["password"]
inst = selectInst(net)
inst["email"] = inst["email"].replace("{nickname}", nickname)
for i in inst.keys():
if not isinstance(inst[i], str):
continue
inst[i] = inst[i].replace("{nickname}", nickname)
inst[i] = inst[i].replace("{password}", password)
inst[i] = inst[i].replace("{email}", inst["email"])
if token:
inst[i] = inst[i].replace("{token}", token)
return inst
def registerAccount(net, num):
debug("Attempting to register: %s - %i" % (net, num))
sinst = substitute(net, num)
if not inst["register"]:
error("Cannot register for %s: function disabled" % (net))
return False
entity = inst["entity"]
email = inst["email"]
cmd = inst["register"]
email = email.replace("{nickname}", nickname)
cmd = cmd.replace("{password}", password)
cmd = cmd.replace("{email}", email)
name = net+str(num)
main.IRCPool[name].msg(entity, cmd)
main.IRCPool[name].msg(sinst["entity"], sinst["registermsg"])
def confirmAccount(net, num, token):
inst = selectInst(net)
entity = inst["entity"]
cmd = inst["confirm"]
cmd = cmd.replace("{token}", token)
sinst = substitute(net, num, token=token)
name = net+str(num)
main.IRCPool[name].msg(entity, cmd)
main.IRCPool[name].msg(sinst["entity"], sinst["confirm"])
enableAuthentication(net, num)
def confirmRegistration(net, num):
def confirmRegistration(net, num, negativepass=None):
obj = main.network[net]
name = net+str(num)
if name in main.IRCPool.keys():
if not negativepass == None:
main.IRCPool[name].regPing(negativepass=negativepass)
return
debug("Relay authenticated: %s - %i" %(net, num))
main.IRCPool[name].authenticated = True
main.IRCPool[name].recheckList()
@ -79,16 +86,27 @@ def enableAuthentication(net, num):
def registerTest(c):
inst = selectInst(c["net"])
name = c["net"]+str(c["num"])
if inst["check"] == False:
return
if "msg" in c.keys() and not c["msg"] == None:
if inst["checktype"] == "msg":
if c["type"] == "query":
if inst["checkmsg"] in c["msg"] and c["nick"] == inst["entity"]:
confirmRegistration(c["net"], c["num"])
return
if inst["negative"]:
if name in main.IRCPool.keys():
if not main.IRCPool[name]._negativePass == True:
if c["type"] == "query" and c["nick"] == inst["entity"]:
if inst["checknegativemsg"] in c["msg"]:
confirmRegistration(c["net"], c["num"], negativepass=False) # Not passed negative check, report back
return
if inst["checkendnegative"] in c["msg"]:
confirmRegistration(c["net"], c["num"], negativepass=True) # Passed the negative check, report back
return
if inst["ping"]:
if inst["checkmsg2"] in c["msg"] and c["nick"] == inst["entity"]:
print("CHECKMSG2 SUCCEEDED", c["num"])
confirmRegistration(c["net"], c["num"])
return
if inst["checktype"] == "msg":
if inst["checkmsg"] in c["msg"]:
confirmRegistration(c["net"], c["num"])
return
elif inst["checktype"] == "mode":