mirror of
https://github.com/Kozea/Radicale.git
synced 2025-06-26 16:45:52 +00:00
Implementing group based collection matching.
Optimize rights evaluation.
This commit is contained in:
parent
2dc0fd29dc
commit
eda8309a04
2 changed files with 35 additions and 16 deletions
|
@ -35,7 +35,7 @@ class Auth(auth.BaseAuth):
|
||||||
_ldap_secret: str
|
_ldap_secret: str
|
||||||
_ldap_filter: str
|
_ldap_filter: str
|
||||||
_ldap_load_groups: bool
|
_ldap_load_groups: bool
|
||||||
_ldap_groups = []
|
_ldap_groups = set
|
||||||
|
|
||||||
def __init__(self, configuration: config.Configuration) -> None:
|
def __init__(self, configuration: config.Configuration) -> None:
|
||||||
super().__init__(configuration)
|
super().__init__(configuration)
|
||||||
|
@ -77,10 +77,12 @@ class Auth(auth.BaseAuth):
|
||||||
conn.protocol_version = 3
|
conn.protocol_version = 3
|
||||||
conn.set_option(ldap.OPT_REFERRALS, 0)
|
conn.set_option(ldap.OPT_REFERRALS, 0)
|
||||||
conn.simple_bind_s(user_dn,password)
|
conn.simple_bind_s(user_dn,password)
|
||||||
|
tmp = []
|
||||||
if self._ldap_load_groups:
|
if self._ldap_load_groups:
|
||||||
self._ldap_groups = []
|
tmp = []
|
||||||
for t in res[0][1]['memberOf']:
|
for t in res[0][1]['memberOf']:
|
||||||
self._ldap_groups.append(t.decode('utf-8').split(',')[0][3:])
|
tmp.append(t.decode('utf-8').split(',')[0][3:])
|
||||||
|
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
|
||||||
|
|
|
@ -44,30 +44,42 @@ from radicale.log import logger
|
||||||
class Rights(rights.BaseRights):
|
class Rights(rights.BaseRights):
|
||||||
|
|
||||||
_filename: str
|
_filename: str
|
||||||
|
_rights_config
|
||||||
|
|
||||||
def __init__(self, configuration: config.Configuration) -> None:
|
def __init__(self, configuration: config.Configuration) -> None:
|
||||||
super().__init__(configuration)
|
super().__init__(configuration)
|
||||||
self._filename = configuration.get("rights", "file")
|
self._filename = configuration.get("rights", "file")
|
||||||
|
_rights_config = configparser.ConfigParser()
|
||||||
|
try:
|
||||||
|
with open(self._filename, "r") as f:
|
||||||
|
_rights_config.read_file(f)
|
||||||
|
except Exception as e:
|
||||||
|
raise RuntimeError("Failed to load rights file %r: %s" %
|
||||||
|
(self._filename, e)) from e
|
||||||
|
|
||||||
def authorization(self, user: str, path: str) -> str:
|
def authorization(self, user: str, path: str) -> str:
|
||||||
user = user or ""
|
user = user or ""
|
||||||
sane_path = pathutils.strip_path(path)
|
sane_path = pathutils.strip_path(path)
|
||||||
# Prevent "regex injection"
|
# Prevent "regex injection"
|
||||||
escaped_user = re.escape(user)
|
escaped_user = re.escape(user)
|
||||||
rights_config = configparser.ConfigParser()
|
|
||||||
|
for section in _rights_config.sections():
|
||||||
|
user_match = False
|
||||||
|
group_match = []
|
||||||
|
collection_match = False
|
||||||
try:
|
try:
|
||||||
with open(self._filename, "r") as f:
|
collection_pattern = _rights_config.get(section, "collection")
|
||||||
rights_config.read_file(f)
|
user_pattern = _rights_config.get(section, "user", fallback = "")
|
||||||
except Exception as e:
|
groups = _rights_config.get(section, "groups", fallback = "").split(",")
|
||||||
raise RuntimeError("Failed to load rights file %r: %s" %
|
|
||||||
(self._filename, e)) from e
|
|
||||||
for section in rights_config.sections():
|
|
||||||
try:
|
try:
|
||||||
user_pattern = rights_config.get(section, "user")
|
group_match = self._auth._ldap_groups & set(groups)
|
||||||
collection_pattern = rights_config.get(section, "collection")
|
except NameError:
|
||||||
|
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)
|
||||||
collection_match = user_match and re.fullmatch(
|
collection_match = re.fullmatch(
|
||||||
collection_pattern.format(
|
collection_pattern.format(
|
||||||
*(re.escape(s) for s in user_match.groups()),
|
*(re.escape(s) for s in user_match.groups()),
|
||||||
user=escaped_user), sane_path)
|
user=escaped_user), sane_path)
|
||||||
|
@ -75,10 +87,15 @@ class Rights(rights.BaseRights):
|
||||||
raise RuntimeError("Error in section %r of rights file %r: "
|
raise RuntimeError("Error in section %r of rights file %r: "
|
||||||
"%s" % (section, self._filename, e)) from e
|
"%s" % (section, self._filename, e)) from e
|
||||||
if user_match and collection_match:
|
if user_match and collection_match:
|
||||||
logger.debug("Rule %r:%r matches %r:%r from section %r",
|
logger.debug("User rule %r:%r matches %r:%r from section %r",
|
||||||
user, sane_path, user_pattern,
|
user, sane_path, user_pattern,
|
||||||
collection_pattern, section)
|
collection_pattern, section)
|
||||||
return rights_config.get(section, "permissions")
|
return _rights_config.get(section, "permissions")
|
||||||
|
if len(group_match) > 0 and collection_match:
|
||||||
|
logger.debug("Group rule %r:%r matches %r from section %r",
|
||||||
|
group_match, sane_path,
|
||||||
|
collection_pattern, section)
|
||||||
|
return _rights_config.get(section, "permissions")
|
||||||
logger.debug("Rule %r:%r doesn't match %r:%r from section %r",
|
logger.debug("Rule %r:%r doesn't match %r:%r from section %r",
|
||||||
user, sane_path, user_pattern, collection_pattern,
|
user, sane_path, user_pattern, collection_pattern,
|
||||||
section)
|
section)
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue