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

Update and integrate Suelta.

Browse Source
This commit is contained in:
Lance Stout
2012-07-29 17:22:16 -07:00
parent e4b4c67637
commit 695cd95657
32 changed files with 859 additions and 1505 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
sleekxmpp/thirdparty/__init__.py vendored
View File

@@ -8,5 +8,4 @@ try:
except:
from sleekxmpp.thirdparty.gnupg import GPG
from sleekxmpp.thirdparty import suelta
from sleekxmpp.thirdparty.mini_dateutil import tzutc, tzoffset, parse_iso

21
sleekxmpp/thirdparty/suelta/LICENSE vendored
View File

@@ -1,21 +0,0 @@
This software is subject to "The MIT License"
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.

27
sleekxmpp/thirdparty/suelta/PLAYING-NICELY vendored
View File

@@ -1,27 +0,0 @@
Hi.
This is a short note explaining the license in non-legally-binding terms, and
describing how I hope to see people work with the licensing.
First off, the license is permissive, and more or less allows you to do
anything, as long as you leave my credit and copyright intact.
You can, and are very much welcome to, include this in commercial works, and
in code that has tightly controlled distribution, as well as open-source.
If it doesn't work - and I have no doubt that there are bugs - then this is
largely your problem.
If you do find a bug, though, do let me know - although you don't have to.
And if you fix it, I'd greatly appreciate a patch, too. Please give me a
licensing statement, and a copyright statement, along with your patch.
Similarly, any enhancements are welcome, and also will need copyright and
licensing. Please stick to a license which is compatible with the MIT license,
and consider assignment (as required) to me to simplify licensing. (Public
domain does not exist in the UK, sorry).
Thanks,
Dave.

8
sleekxmpp/thirdparty/suelta/README vendored
View File

@@ -1,8 +0,0 @@
Suelta - A pure-Python SASL client library
Suelta is a SASL library, providing you with authentication and in some cases
security layers.
It supports a wide range of typical SASL mechanisms, including the MTI for
all known protocols.

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

@@ -1,26 +0,0 @@
# 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)

35
sleekxmpp/thirdparty/suelta/exceptions.py vendored
View File

@@ -1,35 +0,0 @@
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)
class SASLPrepFailure(UnicodeError):
pass

8
sleekxmpp/thirdparty/suelta/mechanisms/__init__.py vendored
View File

@@ -1,8 +0,0 @@
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
from sleekxmpp.thirdparty.suelta.mechanisms.messenger_oauth2 import X_MESSENGER_OAUTH2
from sleekxmpp.thirdparty.suelta.mechanisms.facebook_platform import X_FACEBOOK_PLATFORM
from sleekxmpp.thirdparty.suelta.mechanisms.google_token import X_GOOGLE_TOKEN

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

@@ -1,36 +0,0 @@
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__(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
View File

@@ -1,63 +0,0 @@
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=None):
"""
"""
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)

275
sleekxmpp/thirdparty/suelta/mechanisms/digest_md5.py vendored
View File

@@ -1,275 +0,0 @@
import sys
import random
import hmac
from sleekxmpp.thirdparty.suelta.util import hash, bytes, quote
from sleekxmpp.thirdparty.suelta.util import num_to_bytes, bytes_to_num
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)

43
sleekxmpp/thirdparty/suelta/mechanisms/facebook_platform.py vendored
View File

@@ -1,43 +0,0 @@
from sleekxmpp.thirdparty.suelta.util import bytes
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
try:
import urlparse
except ImportError:
import urllib.parse as urlparse
class X_FACEBOOK_PLATFORM(Mechanism):
def __init__(self, sasl, name):
super(X_FACEBOOK_PLATFORM, self).__init__(sasl, name)
self.check_values(['access_token', 'api_key'])
def process(self, challenge=None):
if challenge is not None:
values = {}
for kv in challenge.split(b'&'):
key, value = kv.split(b'=')
values[key] = value
resp_data = {
'method': values[b'method'],
'v': '1.0',
'call_id': '1.0',
'nonce': values[b'nonce'],
'access_token': self.values['access_token'],
'api_key': self.values['api_key']
}
for k, v in resp_data.items():
resp_data[k] = bytes(v).decode('utf-8')
resp = '&'.join(['%s=%s' % (k, v) for k, v in resp_data.items()])
return bytes(resp)
return b''
def okay(self):
return True
register_mechanism('X-FACEBOOK-PLATFORM', 40, X_FACEBOOK_PLATFORM, use_hashes=False)

22
sleekxmpp/thirdparty/suelta/mechanisms/google_token.py vendored
View File

@@ -1,22 +0,0 @@
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 X_GOOGLE_TOKEN(Mechanism):
def __init__(self, sasl, name):
super(X_GOOGLE_TOKEN, self).__init__(sasl, name)
self.check_values(['email', 'access_token'])
def process(self, challenge=None):
email = bytes(self.values['email'])
token = bytes(self.values['access_token'])
return b'\x00' + email + b'\x00' + token
def okay(self):
return True
register_mechanism('X-GOOGLE-TOKEN', 3, X_GOOGLE_TOKEN, use_hashes=False)

17
sleekxmpp/thirdparty/suelta/mechanisms/messenger_oauth2.py vendored
View File

@@ -1,17 +0,0 @@
from sleekxmpp.thirdparty.suelta.util import bytes
from sleekxmpp.thirdparty.suelta.sasl import Mechanism, register_mechanism
class X_MESSENGER_OAUTH2(Mechanism):
def __init__(self, sasl, name):
super(X_MESSENGER_OAUTH2, self).__init__(sasl, name)
self.check_values(['access_token'])
def process(self, challenge=None):
return bytes(self.values['access_token'])
def okay(self):
return True
register_mechanism('X-MESSENGER-OAUTH2', 10, X_MESSENGER_OAUTH2, use_hashes=False)

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

@@ -1,61 +0,0 @@
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', 5, PLAIN, use_hashes=False)

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

@@ -1,176 +0,0 @@
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(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
View File

@@ -1,402 +0,0 @@
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 in ['', 'anonymous', None]:
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 is not 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']

81
sleekxmpp/thirdparty/suelta/saslprep.py vendored
View File

@@ -1,81 +0,0 @@
from __future__ import unicode_literals
import sys
import stringprep
import unicodedata
from sleekxmpp.thirdparty.suelta.exceptions import SASLPrepFailure
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('utf-8')
# 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 SASLPrepFailure('Section 6.3 [end]')
# Check for prohibited characters
for x in range(len(text)):
if strict and stringprep.in_table_a1(text[x]):
raise SASLPrepFailure('Unassigned Codepoint')
if stringprep.in_table_c12(text[x]):
raise SASLPrepFailure('In table C.1.2')
if stringprep.in_table_c21(text[x]):
raise SASLPrepFailure('In table C.2.1')
if stringprep.in_table_c22(text[x]):
raise SASLPrepFailure('In table C.2.2')
if stringprep.in_table_c3(text[x]):
raise SASLPrepFailure('In table C.3')
if stringprep.in_table_c4(text[x]):
raise SASLPrepFailure('In table C.4')
if stringprep.in_table_c5(text[x]):
raise SASLPrepFailure('In table C.5')
if stringprep.in_table_c6(text[x]):
raise SASLPrepFailure('In table C.6')
if stringprep.in_table_c7(text[x]):
raise SASLPrepFailure('In table C.7')
if stringprep.in_table_c8(text[x]):
raise SASLPrepFailure('In table C.8')
if stringprep.in_table_c9(text[x]):
raise SASLPrepFailure('In table C.9')
if x:
if first_is_randal and stringprep.in_table_d2(text[x]):
raise SASLPrepFailure('Section 6.2')
if not first_is_randal and \
x != len(text) - 1 and \
stringprep.in_table_d1(text[x]):
raise SASLPrepFailure('Section 6.3')
return text

121
sleekxmpp/thirdparty/suelta/util.py vendored
View File

@@ -1,121 +0,0 @@
"""
"""
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 text is None:
return b''
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