mirrors/Radicale
1
0
Fork 0
mirror of https://github.com/Kozea/Radicale.git synced 2025-04-05 06:07:35 +03:00
Code Issues Projects Releases Packages Wiki Activity

Fix the problems found by flake8.

Browse source
This commit is contained in:
Dipl. Ing. Péter Varkoly 2024-09-11 08:12:08 +02:00
parent 5cb16a3a2d
commit d75b071fec
3 changed files with 28 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

2
radicale/app/__init__.py
View file

@ -252,7 +252,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
user = self._auth.login(login, password) or "" if login else "" user = self._auth.login(login, password) or "" if login else ""
if self.configuration.get("auth", "type") == "ldap": if self.configuration.get("auth", "type") == "ldap":
try: try:
logger.debug("Groups %r",",".join(self._auth._ldap_groups)) logger.debug("Groups %r", ",".join(self._auth._ldap_groups))
self._rights._user_groups = self._auth._ldap_groups self._rights._user_groups = self._auth._ldap_groups
except AttributeError: except AttributeError:
pass pass

48
radicale/auth/ldap.py
View file

@ -27,6 +27,7 @@ Following parameters are needed in the configuration
from radicale import auth, config from radicale import auth, config
from radicale.log import logger from radicale.log import logger
class Auth(auth.BaseAuth): class Auth(auth.BaseAuth):
_ldap_uri: str _ldap_uri: str
_ldap_base: str _ldap_base: str
@ -40,33 +41,35 @@ class Auth(auth.BaseAuth):
super().__init__(configuration) super().__init__(configuration)
try: try:
import ldap3 import ldap3
except ImportError as e: self.ldap3 = ldap3
except ImportError:
try: try:
import ldap import ldap
self._ldap_version = 2 self._ldap_version = 2
self.ldap = ldap
except ImportError as e: except ImportError as e:
raise RuntimeError("LDAP authentication requires the ldap3 module") from e raise RuntimeError("LDAP authentication requires the ldap3 module") from e
self._ldap_uri = configuration.get("auth", "ldap_uri") self._ldap_uri = configuration.get("auth", "ldap_uri")
self._ldap_base = configuration.get("auth", "ldap_base") self._ldap_base = configuration.get("auth", "ldap_base")
self._ldap_reader_dn = configuration.get("auth", "ldap_reader_dn") self._ldap_reader_dn = configuration.get("auth", "ldap_reader_dn")
self._ldap_load_groups = configuration.get("auth", "ldap_load_groups") self._ldap_load_groups = configuration.get("auth", "ldap_load_groups")
self._ldap_secret = configuration.get("auth", "ldap_secret") self._ldap_secret = configuration.get("auth", "ldap_secret")
self._ldap_filter = configuration.get("auth", "ldap_filter") self._ldap_filter = configuration.get("auth", "ldap_filter")
def _login2(self, login: str, password: str) -> str: def _login2(self, login: str, password: str) -> str:
try: try:
"""Bind as reader dn""" """Bind as reader dn"""
conn = ldap.initialize(self._ldap_uri) conn = self.ldap.initialize(self._ldap_uri)
conn.protocol_version = 3 conn.protocol_version = 3
conn.set_option(ldap.OPT_REFERRALS, 0) conn.set_option(self.ldap.OPT_REFERRALS, 0)
conn.simple_bind_s(self._ldap_reader_dn, self._ldap_secret) conn.simple_bind_s(self._ldap_reader_dn, self._ldap_secret)
"""Search for the dn of user to authenticate""" """Search for the dn of user to authenticate"""
res = conn.search_s(self._ldap_base, ldap.SCOPE_SUBTREE, filterstr=self._ldap_filter.format(login), attrlist=['memberOf']) res = conn.search_s(self._ldap_base, self.ldap.SCOPE_SUBTREE, filterstr=self._ldap_filter.format(login), attrlist=['memberOf'])
if len(res) == 0: if len(res) == 0:
"""User could not be find""" """User could not be find"""
return "" return ""
user_dn = res[0][0] user_dn = res[0][0]
logger.debug("LDAP Auth user: %s",user_dn) logger.debug("LDAP Auth user: %s", user_dn)
"""Close ldap connection""" """Close ldap connection"""
conn.unbind() conn.unbind()
except Exception: except Exception:
@ -74,27 +77,27 @@ class Auth(auth.BaseAuth):
try: try:
"""Bind as user to authenticate""" """Bind as user to authenticate"""
conn = ldap.initialize(self._ldap_uri) conn = self.ldap.initialize(self._ldap_uri)
conn.protocol_version = 3 conn.protocol_version = 3
conn.set_option(ldap.OPT_REFERRALS, 0) conn.set_option(self.ldap.OPT_REFERRALS, 0)
conn.simple_bind_s(user_dn,password) conn.simple_bind_s(user_dn, password)
tmp = [] tmp = []
if self._ldap_load_groups: if self._ldap_load_groups:
tmp = [] tmp = []
for t in res[0][1]['memberOf']: for t in res[0][1]['memberOf']:
tmp.append(t.decode('utf-8').split(',')[0][3:]) tmp.append(t.decode('utf-8').split(',')[0][3:])
self._ldap_groups = set(tmp) self._ldap_groups = set(tmp)
logger.debug("LDAP Auth groups of user: %s",",".join(self._ldap_groups)) logger.debug("LDAP Auth groups of user: %s", ",".join(self._ldap_groups))
conn.unbind() conn.unbind()
return login return login
except ldap.INVALID_CREDENTIALS: except self.ldap.INVALID_CREDENTIALS:
return "" return ""
def _login3(self, login: str, password: str) -> str: def _login3(self, login: str, password: str) -> str:
"""Connect the server""" """Connect the server"""
try: try:
server = ldap3.Server(self._ldap_uri) server = self.ldap3.Server(self._ldap_uri)
conn = ldap3.Connection(server, self._ldap_reader_dn, password=self._ldap_secret) conn = self.ldap3.Connection(server, self._ldap_reader_dn, password=self._ldap_secret)
except self.ldap3.core.exceptions.LDAPSocketOpenError: except self.ldap3.core.exceptions.LDAPSocketOpenError:
raise RuntimeError("Unable to reach ldap server") raise RuntimeError("Unable to reach ldap server")
except Exception: except Exception:
@ -105,10 +108,10 @@ class Auth(auth.BaseAuth):
"""Search the user dn""" """Search the user dn"""
conn.search( conn.search(
search_base = self._ldap_base, search_base=self._ldap_base,
search_filter = self._ldap_filter.format(login), search_filter=self._ldap_filter.format(login),
search_scope = 'SUBTREE', search_scope='SUBTREE',
attributes = ['memberOf'] attributes=['memberOf']
) )
if len(conn.entries) == 0: if len(conn.entries) == 0:
"""User could not be find""" """User could not be find"""
@ -119,7 +122,7 @@ class Auth(auth.BaseAuth):
user_dn = user_entry['dn'] user_dn = user_entry['dn']
try: try:
"""Try to bind as the user itself""" """Try to bind as the user itself"""
conn = ldap3.Connection(server, user_dn, password=password) conn = self.ldap3.Connection(server, user_dn, password=password)
if not conn.bind(): if not conn.bind():
return "" return ""
if self._ldap_load_groups: if self._ldap_load_groups:
@ -140,6 +143,5 @@ class Auth(auth.BaseAuth):
In the last step the authentication of the user will be proceeded. In the last step the authentication of the user will be proceeded.
""" """
if self._ldap_version == 2: if self._ldap_version == 2:
return _login2(self, login, password) return self._login2(self, login, password)
return _login3(self, login, password) return self._login3(self, login, password)

4
radicale/rights/from_file.py
View file

@ -69,10 +69,10 @@ class Rights(rights.BaseRights):
try: try:
user_pattern = rights_config.get(section, "user") user_pattern = rights_config.get(section, "user")
collection_pattern = rights_config.get(section, "collection") collection_pattern = rights_config.get(section, "collection")
allowed_groups = rights_config.get(section, "groups", fallback = "").split(",") allowed_groups = rights_config.get(section, "groups", fallback="").split(",")
try: try:
group_match = self._user_groups.intersection(allowed_groups) > 0 group_match = self._user_groups.intersection(allowed_groups) > 0
except: except Exception:
pass pass
# Use empty format() for harmonized handling of curly braces # Use empty format() for harmonized handling of curly braces
user_match = re.fullmatch(user_pattern.format(), user) user_match = re.fullmatch(user_pattern.format(), user)