230 lines
7 KiB
Python
230 lines
7 KiB
Python
import ldap
|
|
import logging
|
|
import json
|
|
import itertools
|
|
import uuid
|
|
import struct
|
|
|
|
from util.aes import AESCipher
|
|
from util.validation import generate_valid_usernames
|
|
from data import model
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DatabaseUsers(object):
|
|
def verify_user(self, username_or_email, password):
|
|
""" Simply delegate to the model implementation. """
|
|
return model.verify_user(username_or_email, password)
|
|
|
|
def user_exists(self, username):
|
|
return model.get_user(username) is not None
|
|
|
|
|
|
class LDAPConnection(object):
|
|
def __init__(self, ldap_uri, user_dn, user_pw):
|
|
self._ldap_uri = ldap_uri
|
|
self._user_dn = user_dn
|
|
self._user_pw = user_pw
|
|
self._conn = None
|
|
|
|
def __enter__(self):
|
|
self._conn = ldap.initialize(self._ldap_uri, trace_level=1)
|
|
self._conn.simple_bind_s(self._user_dn, self._user_pw)
|
|
return self._conn
|
|
|
|
def __exit__(self, exc_type, value, tb):
|
|
self._conn.unbind_s()
|
|
|
|
|
|
class LDAPUsers(object):
|
|
def __init__(self, ldap_uri, base_dn, admin_dn, admin_passwd, user_rdn, uid_attr, email_attr):
|
|
self._ldap_conn = LDAPConnection(ldap_uri, admin_dn, admin_passwd)
|
|
self._ldap_uri = ldap_uri
|
|
self._base_dn = base_dn
|
|
self._user_rdn = user_rdn
|
|
self._uid_attr = uid_attr
|
|
self._email_attr = email_attr
|
|
|
|
def _ldap_user_search(self, username_or_email):
|
|
with self._ldap_conn as conn:
|
|
logger.debug('Incoming username or email param: %s', username_or_email.__repr__())
|
|
user_search_dn = ','.join(self._user_rdn + self._base_dn)
|
|
query = u'(|({0}={2})({1}={2}))'.format(self._uid_attr, self._email_attr,
|
|
username_or_email)
|
|
|
|
logger.debug('Conducting user search: %s => %s', user_search_dn, query)
|
|
user = conn.search_s(user_search_dn, ldap.SCOPE_SUBTREE, query.encode('utf-8'))
|
|
|
|
logger.debug('Found user data: %s', user)
|
|
if len(user) != 1:
|
|
return None
|
|
|
|
logger.debug('Found user: %s', user[0])
|
|
return user[0]
|
|
|
|
def verify_user(self, username_or_email, password):
|
|
""" Verify the credentials with LDAP and if they are valid, create or update the user
|
|
in our database. """
|
|
|
|
# Make sure that even if the server supports anonymous binds, we don't allow it
|
|
if not password:
|
|
return None
|
|
|
|
found_user = self._ldap_user_search(username_or_email)
|
|
|
|
if found_user is None:
|
|
return None
|
|
|
|
found_dn, found_response = found_user
|
|
|
|
# First validate the password by binding as the user
|
|
logger.debug('Found user %s; validating password', username_or_email)
|
|
try:
|
|
with LDAPConnection(self._ldap_uri, found_dn, password.encode('utf-8')):
|
|
pass
|
|
except ldap.INVALID_CREDENTIALS:
|
|
logger.exception('Invalid LDAP credentials')
|
|
return None
|
|
|
|
# Now check if we have a federated login for this user
|
|
username = found_response[self._uid_attr][0].decode('utf-8')
|
|
email = found_response[self._email_attr][0]
|
|
db_user = model.verify_federated_login('ldap', username)
|
|
|
|
if not db_user:
|
|
# We must create the user in our db
|
|
valid_username = None
|
|
for valid_username in generate_valid_usernames(username):
|
|
if model.is_username_unique(valid_username):
|
|
break
|
|
|
|
if not valid_username:
|
|
logger.error('Unable to pick a username for user: %s', username)
|
|
return None
|
|
|
|
db_user = model.create_federated_user(valid_username, email, 'ldap', username,
|
|
set_password_notification=False)
|
|
else:
|
|
# Update the db attributes from ldap
|
|
db_user.email = email
|
|
db_user.save()
|
|
|
|
return db_user
|
|
|
|
def user_exists(self, username):
|
|
found_user = self._ldap_user_search(username)
|
|
return found_user is not None
|
|
|
|
|
|
|
|
class UserAuthentication(object):
|
|
def __init__(self, app=None):
|
|
self.app = app
|
|
if app is not None:
|
|
self.state = self.init_app(app)
|
|
else:
|
|
self.state = None
|
|
|
|
def init_app(self, app):
|
|
authentication_type = app.config.get('AUTHENTICATION_TYPE', 'Database')
|
|
|
|
if authentication_type == 'Database':
|
|
users = DatabaseUsers()
|
|
elif authentication_type == 'LDAP':
|
|
ldap_uri = app.config.get('LDAP_URI', 'ldap://localhost')
|
|
base_dn = app.config.get('LDAP_BASE_DN')
|
|
admin_dn = app.config.get('LDAP_ADMIN_DN')
|
|
admin_passwd = app.config.get('LDAP_ADMIN_PASSWD')
|
|
user_rdn = app.config.get('LDAP_USER_RDN', [])
|
|
uid_attr = app.config.get('LDAP_UID_ATTR', 'uid')
|
|
email_attr = app.config.get('LDAP_EMAIL_ATTR', 'mail')
|
|
|
|
users = LDAPUsers(ldap_uri, base_dn, admin_dn, admin_passwd, user_rdn, uid_attr, email_attr)
|
|
|
|
else:
|
|
raise RuntimeError('Unknown authentication type: %s' % authentication_type)
|
|
|
|
# register extension with app
|
|
app.extensions = getattr(app, 'extensions', {})
|
|
app.extensions['authentication'] = users
|
|
return users
|
|
|
|
def _get_secret_key(self):
|
|
""" Returns the secret key to use for encrypting and decrypting. """
|
|
from app import app
|
|
app_secret_key = app.config['SECRET_KEY']
|
|
secret_key = None
|
|
|
|
# First try parsing the key as an int.
|
|
try:
|
|
big_int = int(app_secret_key)
|
|
secret_key = str(bytearray.fromhex('{:02x}'.format(big_int)))
|
|
except ValueError:
|
|
pass
|
|
|
|
# Next try parsing it as an UUID.
|
|
if secret_key is None:
|
|
try:
|
|
secret_key = uuid.UUID(app_secret_key).bytes
|
|
except ValueError:
|
|
pass
|
|
|
|
if secret_key is None:
|
|
secret_key = str(bytearray(map(ord, app_secret_key)))
|
|
|
|
# Otherwise, use the bytes directly.
|
|
return ''.join(itertools.islice(itertools.cycle(secret_key), 32))
|
|
|
|
def encrypt_user_password(self, password):
|
|
""" Returns an encrypted version of the user's password. """
|
|
data = {
|
|
'password': password
|
|
}
|
|
|
|
message = json.dumps(data)
|
|
cipher = AESCipher(self._get_secret_key())
|
|
return cipher.encrypt(message)
|
|
|
|
def _decrypt_user_password(self, encrypted):
|
|
""" Attempts to decrypt the given password and returns it. """
|
|
cipher = AESCipher(self._get_secret_key())
|
|
|
|
try:
|
|
message = cipher.decrypt(encrypted)
|
|
except ValueError:
|
|
return None
|
|
except TypeError:
|
|
return None
|
|
|
|
try:
|
|
data = json.loads(message)
|
|
except ValueError:
|
|
return None
|
|
|
|
return data.get('password', encrypted)
|
|
|
|
def verify_user(self, username_or_email, password, basic_auth=False):
|
|
# First try to decode the password as a signed token.
|
|
if basic_auth:
|
|
import features
|
|
|
|
decrypted = self._decrypt_user_password(password)
|
|
if decrypted is None:
|
|
# This is a normal password.
|
|
if features.REQUIRE_ENCRYPTED_BASIC_AUTH:
|
|
msg = ('Client login with unecrypted passwords is disabled. Please generate an ' +
|
|
'encrypted password in the user admin panel for use here.')
|
|
return (None, msg)
|
|
else:
|
|
password = decrypted
|
|
|
|
result = self.state.verify_user(username_or_email, password)
|
|
if result:
|
|
return (result, '')
|
|
else:
|
|
return (result, 'Invalid password.')
|
|
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self.state, name, None)
|