PapaTutuWawa/etesync-server
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Format code using black

Browse Source
This commit is contained in:
PapaTutuWawa
2020-11-14 16:44:45 +01:00
parent 6751502e21
commit 2d11c82e32
2 changed files with 52 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
myauth/ldap.py
View File

@@ -9,16 +9,17 @@ import ldap
def ldap_setting(name, default):
'''Wrapper around django.conf.settings'''
return getattr(settings, f'LDAP_{name}', default)
"""Wrapper around django.conf.settings"""
return getattr(settings, f"LDAP_{name}", default)
class LDAPConnection:
__instance__ = None
__user_cache = {} # Username -> Valid until
__user_cache = {} # Username -> Valid until
@staticmethod
def get_instance():
'''To get a Singleton'''
"""To get a Singleton"""
if not LDAPConnection.__instance__:
return LDAPConnection()
else:
@@ -26,18 +27,17 @@ class LDAPConnection:
def __init__(self):
# Cache some settings
self.__LDAP_FILTER = ldap_setting('FILTER', '')
self.__LDAP_SEARCH_BASE = ldap_setting('SEARCH_BASE', '')
self.__LDAP_FILTER = ldap_setting("FILTER", "")
self.__LDAP_SEARCH_BASE = ldap_setting("SEARCH_BASE", "")
self.__ldap_connection = ldap.initialize(ldap_setting('SERVER', ''))
self.__ldap_connection = ldap.initialize(ldap_setting("SERVER", ""))
try:
self.__ldap_connection.simple_bind_s(ldap_setting('BIND_DN', ''),
ldap_setting('BIND_PW', ''))
self.__ldap_connection.simple_bind_s(ldap_setting("BIND_DN", ""), ldap_setting("BIND_PW", ""))
except ldap.LDAPError as err:
logging.error(f'LDAP Error occuring during bind: {err.desc}')
logging.error(f"LDAP Error occuring during bind: {err.desc}")
def __is_cache_valid(self, username):
'''Returns True if the cache entry is still valid. Returns False otherwise.'''
"""Returns True if the cache entry is still valid. Returns False otherwise."""
if username in self.__user_cache:
if timezone.now() <= self.__user_cache[username]:
# Cache entry is still valid
@@ -48,26 +48,24 @@ class LDAPConnection:
del self.__user_cache[username]
def has_user(self, username):
'''
"""
Since we don't care about the password and so authentication
another way, all we care about is whether the user exists.
'''
"""
if self.__is_cache_valid(username):
return True
if username in self.__user_cache:
self.__remove_cache(username)
filterstr = self.__LDAP_FILTER.replace('%s', username)
filterstr = self.__LDAP_FILTER.replace("%s", username)
try:
result = self.__ldap_connection.search_s(self.__LDAP_SEARCH_BASE,
ldap.SCOPE_SUBTREE,
filterstr=filterstr)
result = self.__ldap_connection.search_s(self.__LDAP_SEARCH_BASE, ldap.SCOPE_SUBTREE, filterstr=filterstr)
except ldap.NO_RESULTS_RETURNED:
# We handle the specific error first and the the generic error, as
# we may expect ldap.NO_RESULTS_RETURNED, but not any other error
return False
except ldap.LDAPError as err:
logging.error(f'Error occured while performing an LDAP query: {err.desc}')
logging.error(f"Error occured while performing an LDAP query: {err.desc}")
return False
if len(result) == 1:
@@ -75,19 +73,22 @@ class LDAPConnection:
return True
return False
class LDAPUserExists(BasePermission):
'''
"""
A permission check which first checks with the LDAP directory if the user
exists.
'''
"""
def has_permission(self, request, view):
return LDAPConnection.get_instance().has_user(request.user.username)
def create_user(*args, **kwargs):
'''
"""
A create_user function which first checks if the user already exists in the
configured LDAP directory.
'''
if not LDAPConnection.get_instance().has_user(kwargs['username']):
raise PermissionDenied('User not in the LDAP directory.')
"""
if not LDAPConnection.get_instance().has_user(kwargs["username"]):
raise PermissionDenied("User not in the LDAP directory.")
return User.objects.create_user(*args, **kwargs)