jshiffer/slixmpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate a modified version of Dave Cridland's Suelta SASL library.

Browse Source
This commit is contained in:
Lance Stout
2011-08-03 17:00:51 -07:00
parent 9a6eb333e6
commit d4091dbde6
24 changed files with 1419 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
sleekxmpp/thirdparty/__init__.py vendored
View File

@@ -2,3 +2,5 @@ try:
from collections import OrderedDict
except:
from sleekxmpp.thirdparty.ordereddict import OrderedDict
from sleekxmpp.thirdparty import suelta

26
sleekxmpp/thirdparty/suelta/__init__.py vendored Normal file
View File

@@ -0,0 +1,26 @@
# Copyright 2007-2010 David Alan Cridland
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from sleekxmpp.thirdparty.suelta.saslprep import saslprep
from sleekxmpp.thirdparty.suelta.sasl import *
from sleekxmpp.thirdparty.suelta.mechanisms import *
__version__ = '2.0'
__version_info__ = (2, 0, 0)

31
sleekxmpp/thirdparty/suelta/exceptions.py vendored Normal file
View File

@@ -0,0 +1,31 @@
class SASLError(Exception):
def __init__(self, sasl, text, mech=None):
"""
:param sasl: The main `suelta.SASL` object.
:param text: Descpription of the error.
:param mech: Optional reference to the mechanism object.
:type sasl: `suelta.SASL`
"""
self.sasl = sasl
self.text = text
self.mech = mech
def __str__(self):
if self.mech is None:
return 'SASL Error: %s' % self.text
else:
return 'SASL Error (%s): %s' % (self.mech, self.text)
class SASLCancelled(SASLError):
def __init__(self, sasl, mech=None):
"""
:param sasl: The main `suelta.SASL` object.
:param mech: Optional reference to the mechanism object.
:type sasl: `suelta.SASL`
"""
super(SASLCancelled, self).__init__(sasl, "User cancelled", mech)

5
sleekxmpp/thirdparty/suelta/mechanisms/__init__.py vendored Normal file
View File

@@ -0,0 +1,5 @@
from sleekxmpp.thirdparty.suelta.mechanisms.anonymous import ANONYMOUS
from sleekxmpp.thirdparty.suelta.mechanisms.plain import PLAIN
from sleekxmpp.thirdparty.suelta.mechanisms.cram_md5 import CRAM_MD5
from sleekxmpp.thirdparty.suelta.mechanisms.digest_md5 import DIGEST_MD5
from sleekxmpp.thirdparty.suelta.mechanisms.scram_hmac import SCRAM_HMAC

36
sleekxmpp/thirdparty/suelta/mechanisms/anonymous.py vendored Normal file
View File

@@ -0,0 +1,36 @@
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
from sleekxmpp.thirdparty.suelta.exceptions import SASLError, SASLCancelled
class ANONYMOUS(Mechanism):
"""
"""
def __init__(self, sasl, name):
"""
"""
super(ANONYMOUS, self).__init__(self, sasl, name, 0)
def get_values(self):
"""
"""
return {}
def process(self, challenge=None):
"""
"""
return b'Anonymous, Suelta'
def okay(self):
"""
"""
return True
def get_user(self):
"""
"""
return 'anonymous'
register_mechanism('ANONYMOUS', 0, ANONYMOUS, use_hashes=False)

63
sleekxmpp/thirdparty/suelta/mechanisms/cram_md5.py vendored Normal file
View File

@@ -0,0 +1,63 @@
import sys
import hmac
from sleekxmpp.thirdparty.suelta.util import hash, bytes
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
from sleekxmpp.thirdparty.suelta.exceptions import SASLError, SASLCancelled
class CRAM_MD5(Mechanism):
"""
"""
def __init__(self, sasl, name):
"""
"""
super(CRAM_MD5, self).__init__(sasl, name, 2)
self.hash = hash(name[5:])
if self.hash is None:
raise SASLCancelled(self.sasl, self)
if not self.sasl.tls_active():
if not self.sasl.sec_query(self, 'CRAM-MD5'):
raise SASLCancelled(self.sasl, self)
def prep(self):
"""
"""
if 'savepass' not in self.values:
if self.sasl.sec_query(self, 'CLEAR-PASSWORD'):
self.values['savepass'] = True
if 'savepass' not in self.values:
del self.values['password']
def process(self, challenge):
"""
"""
if challenge is None:
return None
self.check_values(['username', 'password'])
username = bytes(self.values['username'])
password = bytes(self.values['password'])
mac = hmac.HMAC(key=password, digestmod=self.hash)
mac.update(challenge)
return username + b' ' + bytes(mac.hexdigest())
def okay(self):
"""
"""
return True
def get_user(self):
"""
"""
return self.values['username']
register_mechanism('CRAM-', 20, CRAM_MD5)

273
sleekxmpp/thirdparty/suelta/mechanisms/digest_md5.py vendored Normal file
View File

@@ -0,0 +1,273 @@
import sys
import random
from sleekxmpp.thirdparty.suelta.util import hash, bytes, quote
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
from sleekxmpp.thirdparty.suelta.exceptions import SASLError, SASLCancelled
def parse_challenge(stuff):
"""
"""
ret = {}
var = b''
val = b''
in_var = True
in_quotes = False
new = False
escaped = False
for c in stuff:
if sys.version_info >= (3, 0):
c = bytes([c])
if in_var:
if c.isspace():
continue
if c == b'=':
in_var = False
new = True
else:
var += c
else:
if new:
if c == b'"':
in_quotes = True
else:
val += c
new = False
elif in_quotes:
if escaped:
escaped = False
val += c
else:
if c == b'\\':
escaped = True
elif c == b'"':
in_quotes = False
else:
val += c
else:
if c == b',':
if var:
ret[var] = val
var = b''
val = b''
in_var = True
else:
val += c
if var:
ret[var] = val
return ret
class DIGEST_MD5(Mechanism):
"""
"""
enc_magic = 'Digest session key to client-to-server signing key magic'
dec_magic = 'Digest session key to server-to-client signing key magic'
def __init__(self, sasl, name):
"""
"""
super(DIGEST_MD5, self).__init__(sasl, name, 3)
self.hash = hash(name[7:])
if self.hash is None:
raise SASLCancelled(self.sasl, self)
if not self.sasl.tls_active():
if not self.sasl.sec_query(self, '-ENCRYPTION, DIGEST-MD5'):
raise SASLCancelled(self.sasl, self)
self._rspauth_okay = False
self._digest_uri = None
self._a1 = None
self._enc_buf = b''
self._enc_key = None
self._enc_seq = 0
self._max_buffer = 65536
self._dec_buf = b''
self._dec_key = None
self._dec_seq = 0
self._qops = [b'auth']
self._qop = b'auth'
def MAC(self, seq, msg, key):
"""
"""
mac = hmac.HMAC(key=key, digestmod=self.hash)
seqnum = num_to_bytes(seq)
mac.update(seqnum)
mac.update(msg)
return mac.digest()[:10] + b'\x00\x01' + seqnum
def encode(self, text):
"""
"""
self._enc_buf += text
def flush(self):
"""
"""
result = b''
# Leave buffer space for the MAC
mbuf = self._max_buffer - 10 - 2 - 4
while self._enc_buf:
msg = self._encbuf[:mbuf]
mac = self.MAC(self._enc_seq, msg, self._enc_key, self.hash)
self._enc_seq += 1
msg += mac
result += num_to_bytes(len(msg)) + msg
self._enc_buf = self._enc_buf[mbuf:]
return result
def decode(self, text):
"""
"""
self._dec_buf += text
result = b''
while len(self._dec_buf) > 4:
num = bytes_to_num(self._dec_buf)
if len(self._dec_buf) < (num + 4):
return result
mac = self._dec_buf[4:4 + num]
self._dec_buf = self._dec_buf[4 + num:]
msg = mac[:-16]
mac_conf = self.MAC(self._dec_mac, msg, self._dec_key)
if mac[-16:] != mac_conf:
self._desc_sec = None
return result
self._dec_seq += 1
result += msg
return result
def response(self):
"""
"""
vitals = ['username']
if not self.has_values(['key_hash']):
vitals.append('password')
self.check_values(vitals)
resp = {}
if 'auth-int' in self._qops:
self._qop = b'auth-int'
resp['qop'] = self._qop
if 'realm' in self.values:
resp['realm'] = quote(self.values['realm'])
resp['username'] = quote(bytes(self.values['username']))
resp['nonce'] = quote(self.values['nonce'])
if self.values['nc']:
self._cnonce = self.values['cnonce']
else:
self._cnonce = bytes('%s' % random.random())[2:]
resp['cnonce'] = quote(self._cnonce)
self.values['nc'] += 1
resp['nc'] = bytes('%08x' % self.values['nc'])
service = bytes(self.sasl.service)
host = bytes(self.sasl.host)
self._digest_uri = service + b'/' + host
resp['digest-uri'] = quote(self._digest_uri)
a2 = b'AUTHENTICATE:' + self._digest_uri
if self._qop != b'auth':
a2 += b':00000000000000000000000000000000'
resp['maxbuf'] = b'16777215' # 2**24-1
resp['response'] = self.gen_hash(a2)
return b','.join([bytes(k) + b'=' + bytes(v) for k, v in resp.items()])
def gen_hash(self, a2):
"""
"""
if not self.has_values(['key_hash']):
key_hash = self.hash()
user = bytes(self.values['username'])
password = bytes(self.values['password'])
realm = bytes(self.values['realm'])
kh = user + b':' + realm + b':' + password
key_hash.update(kh)
self.values['key_hash'] = key_hash.digest()
a1 = self.hash(self.values['key_hash'])
a1h = b':' + self.values['nonce'] + b':' + self._cnonce
a1.update(a1h)
response = self.hash()
self._a1 = a1.digest()
rv = bytes(a1.hexdigest().lower())
rv += b':' + self.values['nonce']
rv += b':' + bytes('%08x' % self.values['nc'])
rv += b':' + self._cnonce
rv += b':' + self._qop
rv += b':' + bytes(self.hash(a2).hexdigest().lower())
response.update(rv)
return bytes(response.hexdigest().lower())
def mutual_auth(self, cmp_hash):
"""
"""
a2 = b':' + self._digest_uri
if self._qop != b'auth':
a2 += b':00000000000000000000000000000000'
if self.gen_hash(a2) == cmp_hash:
self._rspauth_okay = True
def prep(self):
"""
"""
if 'password' in self.values:
del self.values['password']
self.values['cnonce'] = self._cnonce
def process(self, challenge=None):
"""
"""
if challenge is None:
if self.has_values(['username', 'realm', 'nonce', 'key_hash',
'nc', 'cnonce', 'qops']):
self._qops = self.values['qops']
return self.response()
else:
return None
d = parse_challenge(challenge)
if b'rspauth' in d:
self.mutual_auth(d[b'rspauth'])
else:
if b'realm' not in d:
d[b'realm'] = self.sasl.def_realm
for key in ['nonce', 'realm']:
if bytes(key) in d:
self.values[key] = d[bytes(key)]
self.values['nc'] = 0
self._qops = [b'auth']
if b'qop' in d:
self._qops = [x.strip() for x in d[b'qop'].split(b',')]
self.values['qops'] = self._qops
if b'maxbuf' in d:
self._max_buffer = int(d[b'maxbuf'])
return self.response()
def okay(self):
"""
"""
if self._rspauth_okay and self._qop == b'auth-int':
self._enc_key = self.hash(self._a1 + self.enc_magic).digest()
self._dec_key = self.hash(self._a1 + self.dec_magic).digest()
self.encoding = True
return self._rspauth_okay
register_mechanism('DIGEST-', 30, DIGEST_MD5)

61
sleekxmpp/thirdparty/suelta/mechanisms/plain.py vendored Normal file
View File

@@ -0,0 +1,61 @@
import sys
from sleekxmpp.thirdparty.suelta.util import bytes
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
from sleekxmpp.thirdparty.suelta.exceptions import SASLError, SASLCancelled
class PLAIN(Mechanism):
"""
"""
def __init__(self, sasl, name):
"""
"""
super(PLAIN, self).__init__(sasl, name)
if not self.sasl.tls_active():
if not self.sasl.sec_query(self, '-ENCRYPTION, PLAIN'):
raise SASLCancelled(self.sasl, self)
else:
if not self.sasl.sec_query(self, '+ENCRYPTION, PLAIN'):
raise SASLCancelled(self.sasl, self)
self.check_values(['username', 'password'])
def prep(self):
"""
Prepare for processing by deleting the password if
the user has not approved storing it in the clear.
"""
if 'savepass' not in self.values:
if self.sasl.sec_query(self, 'CLEAR-PASSWORD'):
self.values['savepass'] = True
if 'savepass' not in self.values:
del self.values['password']
return True
def process(self, challenge=None):
"""
Process a challenge request and return the response.
:param challenge: A challenge issued by the server that
must be answered for authentication.
"""
user = bytes(self.values['username'])
password = bytes(self.values['password'])
return b'\x00' + user + b'\x00' + password
def okay(self):
"""
Mutual authentication is not supported by PLAIN.
:returns: ``True``
"""
return True
register_mechanism('PLAIN', 1, PLAIN, use_hashes=False)

176
sleekxmpp/thirdparty/suelta/mechanisms/scram_hmac.py vendored Normal file
View File

@@ -0,0 +1,176 @@
import sys
import hmac
import random
from base64 import b64encode, b64decode
from sleekxmpp.thirdparty.suelta.util import hash, bytes, num_to_bytes, bytes_to_num, XOR
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
from sleekxmpp.thirdparty.suelta.exceptions import SASLError, SASLCancelled
def parse_challenge(challenge):
"""
"""
items = {}
for key, value in [item.split(b'=', 1) for item in challenge.split(b',')]:
items[key] = value
return items
class SCRAM_HMAC(Mechanism):
"""
"""
def __init__(self, sasl, name):
"""
"""
super(SCRAM_HMAC, self).__init__(sasl, name, 0)
self._cb = False
if name[-5:] == '-PLUS':
name = name[:-5]
self._cb = True
self.hash = hash(self.name[6:])
if self.hash is None:
raise SASLCancelled(self.sasl, self)
if not self.sasl.tls_active():
if not self.sasl.sec_query(self, '-ENCRYPTION, SCRAM'):
raise SASLCancelled(self.sasl, self)
self._step = 0
self._rspauth = False
def HMAC(self, key, msg):
"""
"""
return hmac.HMAC(key=key, msg=msg, digestmod=self.hash).digest()
def Hi(self, text, salt, iterations):
"""
"""
text = bytes(text)
ui_1 = self.HMAC(text, salt + b'\0\0\0\01')
ui = ui_1
for i in range(iterations - 1):
ui_1 = self.HMAC(text, ui_1)
ui = XOR(ui, ui_1)
return ui
def H(self, text):
"""
"""
return self.hash(text).digest()
def prep(self):
if 'password' in self.values:
del self.values['password']
def process(self, challenge=None):
"""
"""
steps = {
0: self.process_one,
1: self.process_two,
2: self.process_three
}
return steps[self._step](challenge)
def process_one(self, challenge):
"""
"""
vitals = ['username']
if 'SaltedPassword' not in self.values:
vitals.append('password')
if 'Iterations' not in self.values:
vitals.append('password')
self.check_values(vitals)
username = bytes(self.values['username'])
self._step = 1
self._cnonce = bytes(('%s' % random.random())[2:])
self._soup = b'n=' + username + b',r=' + self._cnonce
self._gs2header = b''
if not self.sasl.tls_active():
if self._cb:
self._gs2header = b'p=tls-unique,,'
else:
self._gs2header = b'y,,'
else:
self._gs2header = b'n,,'
return self._gs2header + self._soup
def process_two(self, challenge):
"""
"""
data = parse_challenge(challenge)
self._step = 2
self._soup += b',' + challenge + b','
self._nonce = data[b'r']
self._salt = b64decode(data[b's'])
self._iter = int(data[b'i'])
if self._nonce[:len(self._cnonce)] != self._cnonce:
raise SASLCancelled(self.sasl, self)
cbdata = self.sasl.tls_active()
c = self._gs2header
if not cbdata and self._cb:
c += None
r = b'c=' + b64encode(c).replace(b'\n', b'')
r += b',r=' + self._nonce
self._soup += r
if 'Iterations' in self.values:
if self.values['Iterations'] != self._iter:
if 'SaltedPassword' in self.values:
del self.values['SaltedPassword']
if 'Salt' in self.values:
if self.values['Salt'] != self._salt:
if 'SaltedPassword' in self.values:
del self.values['SaltedPassword']
self.values['Iterations'] = self._iter
self.values['Salt'] = self._salt
if 'SaltedPassword' not in self.values:
self.check_values(['password'])
password = bytes(self.values['password'])
salted_pass = self.Hi(password, self._salt, self._iter)
self.values['SaltedPassword'] = salted_pass
salted_pass = self.values['SaltedPassword']
client_key = self.HMAC(salted_pass, b'Client Key')
stored_key = self.H(client_key)
client_sig = self.HMAC(stored_key, self._soup)
client_proof = XOR(client_key, client_sig)
r += b',p=' + b64encode(client_proof).replace(b'\n', b'')
server_key = self.HMAC(self.values['SaltedPassword'], b'Server Key')
self.server_sig = self.HMAC(server_key, self._soup)
return r
def process_three(self, challenge=None):
"""
"""
data = parse_challenge(challenge)
if b64decode(data[b'v']) == self.server_sig:
self._rspauth = True
def okay(self):
"""
"""
return self._rspauth
def get_user(self):
return self.values['username']
register_mechanism('SCRAM-', 60, SCRAM_HMAC)
register_mechanism('SCRAM-', 70, SCRAM_HMAC, extra='-PLUS')

402
sleekxmpp/thirdparty/suelta/sasl.py vendored Normal file
View File

@@ -0,0 +1,402 @@
from sleekxmpp.thirdparty.suelta.util import hashes
from sleekxmpp.thirdparty.suelta.saslprep import saslprep
#: Global session storage for user answers to requested mechanism values
#: and security questions. This allows the user's preferences to be
#: persisted across multiple SASL authentication attempts made by the
#: same process.
SESSION = {'answers': {},
'passwords': {},
'sec_queries': {},
'stash': {},
'stash_file': ''}
#: Global registry mapping mechanism names to implementation classes.
MECHANISMS = {}
#: Global registry mapping mechanism names to security scores.
MECH_SEC_SCORES = {}
def register_mechanism(basename, basescore, impl, extra=None, use_hashes=True):
"""
Add a SASL mechanism to the registry of available mechanisms.
:param basename: The base name of the mechanism type, such as ``CRAM-``.
:param basescore: The base security score for this type of mechanism.
:param impl: The class implementing the mechanism.
:param extra: Any additional qualifiers to the mechanism name,
such as ``-PLUS``.
:param use_hashes: If ``True``, then register the mechanism for use with
all available hashes.
"""
n = 0
if use_hashes:
for hashing_alg in hashes():
n += 1
name = basename + hashing_alg
if extra is not None:
name += extra
MECHANISMS[name] = impl
MECH_SEC_SCORES[name] = basescore + n
else:
MECHANISMS[basename] = impl
MECH_SEC_SCORES[basename] = basescore
def set_stash_file(filename):
"""
Enable or disable storing the stash to disk.
If the filename is ``None``, then disable using a stash file.
:param filename: The path to the file to store the stash data.
"""
SESSION['stash_file'] = filename
try:
import marshal
stash_file = file(filename)
SESSION['stash'] = marshal.load(stash_file)
except:
SESSION['stash'] = {}
def sec_query_allow(mech, query):
"""
Quick default to allow all feature combinations which could
negatively affect security.
:param mech: The chosen SASL mechanism
:param query: An encoding of the combination of enabled and
disabled features which may affect security.
:returns: ``True``
"""
return True
class SASL(object):
"""
"""
def __init__(self, host, service, mech=None, username=None,
min_sec=0, request_values=None, sec_query=None,
tls_active=None, def_realm=None):
"""
:param string host: The host of the service requiring authentication.
:param string service: The name of the underlying protocol in use.
:param string mech: Optional name of the SASL mechanism to use.
If given, only this mechanism may be used for
authentication.
:param string username: The username to use when authenticating.
:param request_values: Reference to a function for supplying
values requested by mechanisms, such
as passwords. (See above)
:param sec_query: Reference to a function for approving or
denying feature combinations which could
negatively impact security. (See above)
:param tls_active: Function for indicating if TLS has been
negotiated. (See above)
:param integer min_sec: The minimum security level accepted. This
only allows for SASL mechanisms whose
security rating is greater than `min_sec`.
:param string def_realm: The default realm, if different than `host`.
:type request_values: :func:`request_values`
:type sec_query: :func:`sec_query`
:type tls_active: :func:`tls_active`
"""
self.host = host
self.def_realm = def_realm or host
self.service = service
self.user = username
self.mech = mech
self.min_sec = min_sec - 1
self.request_values = request_values
self._sec_query = sec_query
if tls_active is not None:
self.tls_active = tls_active
else:
self.tls_active = lambda: False
self.try_username = self.user
self.try_password = None
self.stash_id = None
self.testkey = None
def reset_stash_id(self, username):
"""
Reset the ID for the stash for persisting user data.
:param username: The username to base the new ID on.
"""
username = saslprep(username)
self.user = username
self.try_username = self.user
self.testkey = [self.user, self.host, self.service]
self.stash_id = '\0'.join(self.testkey)
def sec_query(self, mech, query):
"""
Request authorization from the user to use a combination
of features which could negatively affect security.
The ``sec_query`` callback when creating the SASL object will
be called if the query has not been answered before. Otherwise,
the query response will be pulled from ``SESSION['sec_queries']``.
If no ``sec_query`` callback was provided, then all queries
will be denied.
:param mech: The chosen SASL mechanism
:param query: An encoding of the combination of enabled and
disabled features which may affect security.
:rtype: bool
"""
if self._sec_query is None:
return False
if query in SESSION['sec_queries']:
return SESSION['sec_queries'][query]
resp = self._sec_query(mech, query)
if resp:
SESSION['sec_queries'][query] = resp
return resp
def find_password(self, mech):
"""
Find and return the user's password, if it has been entered before
during this session.
:param mech: The chosen SASL mechanism.
"""
if self.try_password is not None:
return self.try_password
if self.testkey is None:
return
testkey = self.testkey[:]
lockout = 1
def find_username(self):
"""Find and return user's username if known."""
return self.try_username
def success(self, mech):
mech.preprep()
if 'password' in mech.values:
testkey = self.testkey[:]
while len(testkey):
tk = '\0'.join(testkey)
if tk in SESSION['passwords']:
break
SESSION['passwords'][tk] = mech.values['password']
testkey = testkey[:-1]
mech.prep()
mech.save_values()
def failure(self, mech):
mech.clear()
self.testkey = self.testkey[:-1]
def choose_mechanism(self, mechs, force_plain=False):
"""
Choose the most secure mechanism from a list of mechanisms.
If ``force_plain`` is given, return the ``PLAIN`` mechanism.
:param mechs: A list of mechanism names.
:param force_plain: If ``True``, force the selection of the
``PLAIN`` mechanism.
:returns: A SASL mechanism object, or ``None`` if no mechanism
could be selected.
"""
# Handle selection of PLAIN and ANONYMOUS
if force_plain:
return MECHANISMS['PLAIN'](self, 'PLAIN')
if self.user is not None:
requested_mech = '*' if self.mech is None else self.mech
else:
if self.mech is None:
requested_mech = 'ANONYMOUS'
else:
requested_mech = self.mech
if requested_mech == '*' and self.user == 'anonymous':
requested_mech = 'ANONYMOUS'
# If a specific mechanism was requested, try it
if requested_mech != '*':
if requested_mech in MECHANISMS and \
requested_mech in MECH_SEC_SCORES:
return MECHANISMS[requested_mech](self, requested_mech)
return None
# Pick the best mechanism based on its security score
best_score = self.min_sec
best_mech = None
for name in mechs:
if name in MECH_SEC_SCORES:
if MECH_SEC_SCORES[name] > best_score:
best_score = MECH_SEC_SCORES[name]
best_mech = name
if best_mech != None:
best_mech = MECHANISMS[best_mech](self, best_mech)
return best_mech
class Mechanism(object):
"""
"""
def __init__(self, sasl, name, version=0, use_stash=True):
self.name = name
self.sasl = sasl
self.use_stash = use_stash
self.encoding = False
self.values = {}
if use_stash:
self.load_values()
def load_values(self):
"""Retrieve user data from the stash."""
self.values = {}
if not self.use_stash:
return False
if self.sasl.stash_id is not None:
if self.sasl.stash_id in SESSION['stash']:
if SESSION['stash'][self.sasl.stash_id]['mech'] == self.name:
values = SESSION['stash'][self.sasl.stash_id]['values']
self.values.update(values)
if self.sasl.user is not None:
if not self.has_values(['username']):
self.values['username'] = self.sasl.user
return None
def save_values(self):
"""
Save user data to the session stash.
If a stash file name has been set using ``SESSION['stash_file']``,
the saved values will be persisted to disk.
"""
if not self.use_stash:
return False
if self.sasl.stash_id is not None:
if self.sasl.stash_id not in SESSION['stash']:
SESSION['stash'][self.sasl.stash_id] = {}
SESSION['stash'][self.sasl.stash_id]['values'] = self.values
SESSION['stash'][self.sasl.stash_id]['mech'] = self.name
if SESSION['stash_file'] not in ['', None]:
import marshal
stash_file = file(SESSION['stash_file'], 'wb')
marshal.dump(SESSION['stash'], stash_file)
def clear(self):
"""Reset all user data, except the username."""
username = None
if 'username' in self.values:
username = self.values['username']
self.values = {}
if username is not None:
self.values['username'] = username
self.save_values()
self.values = {}
self.load_values()
def okay(self):
"""
Indicate if mutual authentication has completed successfully.
:rtype: bool
"""
return False
def preprep(self):
"""Ensure that the stash ID has been set before processing."""
if self.sasl.stash_id is None:
if 'username' in self.values:
self.sasl.reset_stash_id(self.values['username'])
def prep(self):
"""
Prepare stored values for processing.
For example, by removing extra copies of passwords from memory.
"""
pass
def process(self, challenge=None):
"""
Process a challenge request and return the response.
:param challenge: A challenge issued by the server that
must be answered for authentication.
"""
raise NotImplemented
def fulfill(self, values):
"""
Provide requested values to the mechanism.
:param values: A dictionary of requested values.
"""
if 'password' in values:
values['password'] = saslprep(values['password'])
self.values.update(values)
def missing_values(self, keys):
"""
Return a dictionary of value names that have not been given values
by the user, or retrieved from the stash.
:param keys: A list of value names to check.
:rtype: dict
"""
vals = {}
for name in keys:
if name not in self.values or self.values[name] is None:
if self.use_stash:
if name == 'username':
value = self.sasl.find_username()
if value is not None:
self.sasl.reset_stash_id(value)
self.values[name] = value
break
if name == 'password':
value = self.sasl.find_password(self)
if value is not None:
self.values[name] = value
break
vals[name] = None
return vals
def has_values(self, keys):
"""
Check that the given values have been retrieved from the user,
or from the stash.
:param keys: A list of value names to check.
"""
return len(self.missing_values(keys)) == 0
def check_values(self, keys):
"""
Request missing values from the user.
:param keys: A list of value names to request, if missing.
"""
vals = self.missing_values(keys)
if vals:
self.sasl.request_values(self, vals)
def get_user(self):
"""Return the username usd for this mechanism."""
return self.values['username']

78
sleekxmpp/thirdparty/suelta/saslprep.py vendored Normal file
View File

@@ -0,0 +1,78 @@
from __future__ import unicode_literals
import sys
import stringprep
import unicodedata
def saslprep(text, strict=True):
"""
Return a processed version of the given string, using the SASLPrep
profile of stringprep.
:param text: The string to process, in UTF-8.
:param strict: If ``True``, prevent the use of unassigned code points.
"""
if sys.version_info < (3, 0):
if type(text) == str:
text = text.decode('us-ascii')
# Mapping:
#
# - non-ASCII space characters [StringPrep, C.1.2] that can be
# mapped to SPACE (U+0020), and
#
# - the 'commonly mapped to nothing' characters [StringPrep, B.1]
# that can be mapped to nothing.
buffer = ''
for char in text:
if stringprep.in_table_c12(char):
buffer += ' '
elif not stringprep.in_table_b1(char):
buffer += char
# Normalization using form KC
text = unicodedata.normalize('NFKC', buffer)
# Check for bidirectional string
buffer = ''
first_is_randal = False
if text:
first_is_randal = stringprep.in_table_d1(text[0])
if first_is_randal and not stringprep.in_table_d1(text[-1]):
raise UnicodeError('Section 6.3 [end]')
# Check for prohibited characters
for x in range(len(text)):
if strict and stringprep.in_table_a1(text[x]):
raise UnicodeError('Unassigned Codepoint')
if stringprep.in_table_c12(text[x]):
raise UnicodeError('In table C.1.2')
if stringprep.in_table_c21(text[x]):
raise UnicodeError('In table C.2.1')
if stringprep.in_table_c22(text[x]):
raise UnicodeError('In table C.2.2')
if stringprep.in_table_c3(text[x]):
raise UnicodeError('In table C.3')
if stringprep.in_table_c4(text[x]):
raise UnicodeError('In table C.4')
if stringprep.in_table_c5(text[x]):
raise UnicodeError('In table C.5')
if stringprep.in_table_c6(text[x]):
raise UnicodeError('In table C.6')
if stringprep.in_table_c7(text[x]):
raise UnicodeError('In table C.7')
if stringprep.in_table_c8(text[x]):
raise UnicodeError('In table C.8')
if stringprep.in_table_c9(text[x]):
raise UnicodeError('In table C.9')
if x:
if first_is_randal and stringprep.in_table_d2(text[x]):
raise UnicodeError('Section 6.2')
if not first_is_randal and \
x != len(text) - 1 and \
stringprep.in_table_d1(text[x]):
raise UnicodeError('Section 6.3')
return text

118
sleekxmpp/thirdparty/suelta/util.py vendored Normal file
View File

@@ -0,0 +1,118 @@
"""
"""
import sys
import hashlib
def bytes(text):
"""
Convert Unicode text to UTF-8 encoded bytes.
Since Python 2.6+ and Python 3+ have similar but incompatible
signatures, this function unifies the two to keep code sane.
:param text: Unicode text to convert to bytes
:rtype: bytes (Python3), str (Python2.6+)
"""
if sys.version_info < (3, 0):
import __builtin__
return __builtin__.bytes(text)
else:
import builtins
if isinstance(text, builtins.bytes):
# We already have bytes, so do nothing
return text
if isinstance(text, list):
# Convert a list of integers to bytes
return builtins.bytes(text)
else:
# Convert UTF-8 text to bytes
return builtins.bytes(text, encoding='utf-8')
def quote(text):
"""
Enclose in quotes and escape internal slashes and double quotes.
:param text: A Unicode or byte string.
"""
text = bytes(text)
return b'"' + text.replace(b'\\', b'\\\\').replace(b'"', b'\\"') + b'"'
def num_to_bytes(num):
"""
Convert an integer into a four byte sequence.
:param integer num: An integer to convert to its byte representation.
"""
bval = b''
bval += bytes(chr(0xFF & (num >> 24)))
bval += bytes(chr(0xFF & (num >> 16)))
bval += bytes(chr(0xFF & (num >> 8)))
bval += bytes(chr(0xFF & (num >> 0)))
return bval
def bytes_to_num(bval):
"""
Convert a four byte sequence to an integer.
:param bytes bval: A four byte sequence to turn into an integer.
"""
num = 0
num += ord(bval[0] << 24)
num += ord(bval[1] << 16)
num += ord(bval[2] << 8)
num += ord(bval[3])
return num
def XOR(x, y):
"""
Return the results of an XOR operation on two equal length byte strings.
:param bytes x: A byte string
:param bytes y: A byte string
:rtype: bytes
"""
result = b''
for a, b in zip(x, y):
if sys.version_info < (3, 0):
result += chr((ord(a) ^ ord(b)))
else:
result += bytes([a ^ b])
return result
def hash(name):
"""
Return a hash function implementing the given algorithm.
:param name: The name of the hashing algorithm to use.
:type name: string
:rtype: function
"""
name = name.lower()
if name.startswith('sha-'):
name = 'sha' + name[4:]
if name in dir(hashlib):
return getattr(hashlib, name)
return None
def hashes():
"""
Return a list of available hashing algorithms.
:rtype: list of strings
"""
t = []
if 'md5' in dir(hashlib):
t = ['MD5']
if 'md2' in dir(hashlib):
t += ['MD2']
hashes = ['SHA-' + h[3:] for h in dir(hashlib) if h.startswith('sha')]
return t + hashes