Mikaela
/
Limnoria
mirror of https://github.com/Mikaela/Limnoria.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Try all possible SASL mechanisms instead of just one.

This commit is contained in:
Valentin Lorentz 2015-12-11 10:56:05 +01:00
parent a72926ad11
commit 45c23a8f54
3 changed files with 183 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/conf.py
View File

@ -300,6 +300,12 @@ class SpaceSeparatedSetOfChannels(registry.SpaceSeparatedListOf):
# Let's be explicit about it
return None
class ValidSaslMechanism(registry.OnlySomeStrings):
validStrings = ('ecdsa-nist256p-challenge', 'external', 'plain')
class SpaceSeparatedListOfSaslMechanisms(registry.SpaceSeparatedListOf):
Value = ValidSaslMechanism
def registerNetwork(name, password='', ssl=False, sasl_username='',
sasl_password=''):
network = registerGroup(supybot.networks, name)
@ -348,6 +354,9 @@ def registerNetwork(name, password='', ssl=False, sasl_username='',
_("""Determines what SASL ECDSA key (if any) will be used on %s.
The public key must be registered with NickServ for SASL
ECDSA-NIST256P-CHALLENGE to work.""") % name, private=False))
registerGlobalValue(sasl, 'mechanisms', SpaceSeparatedListOfSaslMechanisms(
['ecdsa-nist256p-challenge', 'external', 'plain'], _("""Determines
what SASL mechanisms will be tried and in which order.""")))
registerGlobalValue(network, 'socksproxy', registry.String('',
_("""If not empty, determines the hostname of the socks proxy that
will be used to connect to this network.""")))

148
src/irclib.py
View File

@ -982,6 +982,22 @@ class Irc(IrcCommandDispatcher, log.Firewalled):
self.lastping = time.time()
self.outstandingPing = False
network_config = conf.supybot.networks.get(self.network)
self.sasl_next_mechanisms = []
self.sasl_current_mechanism = None
for mechanism in network_config.sasl.mechanisms():
if mechanism == 'ecdsa-nist256p-challenge' and \
ecdsa and self.sasl_username and self.sasl_ecdsa_key:
self.sasl_next_mechanisms.append(mechanism)
elif mechanism == 'external' and (
network_config.certfile() or
conf.supybot.protocols.irc.certfile()):
self.sasl_next_mechanisms.append(mechanism)
elif mechanism == 'plain' and \
self.sasl_username and self.sasl_password:
self.sasl_next_mechanisms.append(mechanism)
REQUEST_CAPABILITIES = set(['account-notify', 'extended-join',
'multi-prefix', 'metadata-notify', 'account-tag',
@ -1016,19 +1032,23 @@ class Irc(IrcCommandDispatcher, log.Firewalled):
self.sendMsg(ircmsgs.user(self.ident, self.user))
self.sasl = None
if ecdsa and self.sasl_username and self.sasl_ecdsa_key:
self.sasl = 'ecdsa-nist256p-challenge'
elif (conf.supybot.networks.get(self.network).certfile() or
conf.supybot.protocols.irc.certfile()):
self.sasl = 'external'
elif self.sasl_username and self.sasl_password:
self.sasl = 'plain'
if self.sasl:
if self.sasl_next_mechanisms:
self.REQUEST_CAPABILITIES.add('sasl')
def sendSaslString(self, string):
for chunk in ircutils.authenticate_generator(string):
self.sendMsg(ircmsgs.IrcMsg(command='AUTHENTICATE',
args=(chunk,)))
def tryNextSaslMechanism(self):
if self.sasl_next_mechanisms:
self.sasl_current_mechanism = self.sasl_next_mechanisms.pop(0)
self.sendMsg(ircmsgs.IrcMsg(command='AUTHENTICATE',
args=(self.sasl_current_mechanism.upper(),)))
else:
self.sasl_current_mechanism = None
self.sendMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def doAuthenticate(self, msg):
if not self.authenticate_decoder:
self.authenticate_decoder = ircutils.AuthenticateDecoder()
@ -1037,34 +1057,57 @@ class Irc(IrcCommandDispatcher, log.Firewalled):
return # Waiting for other messages
string = self.authenticate_decoder.get()
self.authenticate_decoder = None
if string == b'':
log.info('%s: Authenticating using SASL.', self.network)
if self.sasl == 'external':
authstring = b''
elif self.sasl == 'ecdsa-nist256p-challenge':
authstring = self.sasl_username.encode('utf-8')
elif self.sasl == 'plain':
authstring = b'\0'.join([
self.sasl_username.encode('utf-8'),
self.sasl_username.encode('utf-8'),
self.sasl_password.encode('utf-8'),
])
for chunk in ircutils.authenticate_generator(authstring):
self.sendMsg(ircmsgs.IrcMsg(command='AUTHENTICATE',
args=(chunk,)))
elif (string != b'' and self.sasl == 'ecdsa-nist256p-challenge'):
mechanism = self.sasl_current_mechanism
if mechanism == 'ecdsa-nist256p-challenge':
if string == b'':
self.sendSaslString(self.sasl_username.encode('utf-8'))
return
try:
with open(self.sasl_ecdsa_key) as fd:
private_key = SigningKey.from_pem(fd.read())
authstring = private_key.sign(base64.b64decode(msg.args[0].encode()))
chunks = ircutils.authenticate_generator(authstring)
self.sendSaslString(authstring)
except (BadDigestError, OSError, ValueError):
chunks = ['*']
for chunk in chunks:
self.sendMsg(ircmsgs.IrcMsg(command='AUTHENTICATE',
args=(chunk,)))
args=('*',)))
self.tryNextSaslMechanism()
elif mechanism == 'external':
self.sendSaslString(b'')
elif mechanism == 'plain':
authstring = b'\0'.join([
self.sasl_username.encode('utf-8'),
self.sasl_username.encode('utf-8'),
self.sasl_password.encode('utf-8'),
])
self.sendSaslString(authstring)
def do903(self, msg):
log.info('%s: SASL authentication successful', self.network)
self.queueMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def do904(self, msg):
log.warning('%s: SASL authentication failed', self.network)
self.tryNextSaslMechanism()
def do905(self, msg):
log.warning('%s: SASL authentication failed because the username or '
'password is too long.', self.network)
self.tryNextSaslMechanism()
def do906(self, msg):
log.warning('%s: SASL authentication aborted', self.network)
self.tryNextSaslMechanism()
def do907(self, msg):
log.warning('%s: Attempted SASL authentication when we were already '
'authenticated.', self.network)
self.tryNextSaslMechanism()
def do908(self, msg):
log.info('%s: Supported SASL mechanisms: %s',
self.network, msg.args[1])
# TODO: filter self.sasl_next_mechanisms
def doCap(self, msg):
subcommand = msg.args[1]
@ -1086,8 +1129,8 @@ class Irc(IrcCommandDispatcher, log.Firewalled):
self.network, caps)
self.state.capabilities_ack.update(caps)
if 'sasl' in caps and self.sasl:
self.sendMsg(ircmsgs.IrcMsg(command='AUTHENTICATE', args=(self.sasl.upper(),)))
if 'sasl' in caps:
self.tryNextSaslMechanism()
else:
self.sendMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def doCapNak(self, msg):
@ -1110,6 +1153,7 @@ class Irc(IrcCommandDispatcher, log.Firewalled):
else:
self.state.capabilities_ls[item] = None
def doCapLs(self, msg):
# TODO: filter self.sasl_next_mechanisms
if len(msg.args) == 4:
# Multi-line LS
if msg.args[2] != '*':
@ -1184,42 +1228,6 @@ class Irc(IrcCommandDispatcher, log.Firewalled):
self.queueMsg(ircmsgs.monitor('-', should_be_unmonitored))
return should_be_unmonitored
def do903(self, msg):
log.info('%s: SASL authentication successful', self.network)
self.queueMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def do904(self, msg):
if (self.sasl != 'plain' and self.sasl_username and
self.sasl_password):
log.info('%s: SASL %s failed, trying PLAIN.', self.network,
self.sasl.upper())
self.sasl = 'plain'
self.queueMsg(ircmsgs.IrcMsg(
command='AUTHENTICATE', args=(self.sasl.upper(),)))
else:
log.warning('%s: SASL authentication failed', self.network)
self.queueMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def do905(self, msg):
log.warning('%s: SASL authentication failed because the username or '
'password is too long.', self.network)
self.queueMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def do906(self, msg):
log.warning('%s: SASL authentication aborted', self.network)
self.queueMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def do907(self, msg):
log.warning('%s: Attempted SASL authentication when we were already '
'authenticated.', self.network)
self.queueMsg(ircmsgs.IrcMsg(command='CAP', args=('END',)))
def do908(self, msg):
log.info('%s: Supported SASL mechanisms: %s',
self.network, msg.args[1])
def _getNextNick(self):
if self.alternateNicks:
nick = self.alternateNicks.pop(0)

96
test/test_irclib.py
View File

@ -527,6 +527,102 @@ class IrcTestCase(SupyTestCase):
self.irc.removeCallback(c.name())
self.assertEqual(c.batch, irclib.Batch('netjoin', (), [m1, m2]))
class SaslTestCase(SupyTestCase):
def setUp(self):
pass
def startCapNegociation(self):
m = self.irc.takeMsg()
self.failUnless(m.command == 'CAP', 'Expected CAP, got %r.' % m)
self.failUnless(m.args == ('LS', '302'), 'Expected CAP LS 302, got %r.' % m)
m = self.irc.takeMsg()
self.failUnless(m.command == 'NICK', 'Expected NICK, got %r.' % m)
m = self.irc.takeMsg()
self.failUnless(m.command == 'USER', 'Expected USER, got %r.' % m)
# TODO
self.irc.feedMsg(ircmsgs.IrcMsg(command='CAP',
args=('*', 'LS', 'sasl')))
m = self.irc.takeMsg()
self.failUnless(m.command == 'CAP', 'Expected CAP, got %r.' % m)
self.assertEqual(m.args[0], 'REQ', m)
self.assertEqual(m.args[1], 'sasl')
self.irc.feedMsg(ircmsgs.IrcMsg(command='CAP',
args=('*', 'ACK', 'sasl')))
def endCapNegociation(self):
m = self.irc.takeMsg()
self.failUnless(m.command == 'CAP', 'Expected CAP, got %r.' % m)
self.assertEqual(m.args, ('END',), m)
def testPlain(self):
try:
conf.supybot.networks.test.sasl.username.setValue('jilles')
conf.supybot.networks.test.sasl.password.setValue('sesame')
self.irc = irclib.Irc('test')
finally:
conf.supybot.networks.test.sasl.username.setValue('')
conf.supybot.networks.test.sasl.password.setValue('')
self.assertEqual(self.irc.sasl_current_mechanism, None)
self.assertEqual(self.irc.sasl_next_mechanisms, ['plain'])
self.startCapNegociation()
m = self.irc.takeMsg()
self.assertEqual(m, ircmsgs.IrcMsg(command='AUTHENTICATE',
args=('PLAIN',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='AUTHENTICATE', args=('+',)))
m = self.irc.takeMsg()
self.assertEqual(m, ircmsgs.IrcMsg(command='AUTHENTICATE',
args=('amlsbGVzAGppbGxlcwBzZXNhbWU=',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='900', args=('jilles',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='903', args=('jilles',)))
self.endCapNegociation()
def testEcdsaFallbackToPlain(self):
try:
conf.supybot.networks.test.sasl.username.setValue('jilles')
conf.supybot.networks.test.sasl.password.setValue('sesame')
conf.supybot.networks.test.sasl.ecdsa_key.setValue('foo')
self.irc = irclib.Irc('test')
finally:
conf.supybot.networks.test.sasl.username.setValue('')
conf.supybot.networks.test.sasl.password.setValue('')
conf.supybot.networks.test.sasl.ecdsa_key.setValue('')
self.assertEqual(self.irc.sasl_current_mechanism, None)
self.assertEqual(self.irc.sasl_next_mechanisms,
['ecdsa-nist256p-challenge', 'plain'])
self.startCapNegociation()
m = self.irc.takeMsg()
self.assertEqual(m, ircmsgs.IrcMsg(command='AUTHENTICATE',
args=('ECDSA-NIST256P-CHALLENGE',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='904',
args=('mechanism not available',)))
m = self.irc.takeMsg()
self.assertEqual(m, ircmsgs.IrcMsg(command='AUTHENTICATE',
args=('PLAIN',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='AUTHENTICATE', args=('+',)))
m = self.irc.takeMsg()
self.assertEqual(m, ircmsgs.IrcMsg(command='AUTHENTICATE',
args=('amlsbGVzAGppbGxlcwBzZXNhbWU=',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='900', args=('jilles',)))
self.irc.feedMsg(ircmsgs.IrcMsg(command='903', args=('jilles',)))
self.endCapNegociation()