oss/Radicale
oss/Radicale
1
0
Fork 0
mirror of https://github.com/Kozea/Radicale.git synced 2025-06-26 16:45:52 +00:00
Code Wiki Activity

Merge pull request #1576 from petervarkoly/master

Browse source 
Implementing group calendars and increase perfomance
This commit is contained in:
Peter Bieringer 2024-09-22 20:42:05 +02:00 • committed by GitHub
commit fdb014d068
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 46 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

6
config
View file

@ -59,7 +59,7 @@
# URI to the LDAP server
#ldap_uri = ldap://localhost
# The base DN of the LDAP server
# The base DN where the user accounts have to be searched
#ldap_base = ##BASE_DN##
# The reader DN of the LDAP server
@ -71,8 +71,8 @@
# If the ldap groups of the user need to be loaded
#ldap_load_groups = True
# Value: none | htpasswd | remote_user | http_x_remote_user | denyall
#type = none
# The filter to find the DN of the user. This filter must contain a python-style placeholder for the login
#ldap_filter = (&(objectClass=person)(cn={0}))
# Htpasswd filename
#htpasswd_filename = /etc/radicale/users

3
radicale/app/propfind.py
View file

@ -392,7 +392,8 @@ class ApplicationPartPropfind(ApplicationBase):
return httputils.REQUEST_TIMEOUT
with self._storage.acquire_lock("r", user):
items_iter = iter(self._storage.discover(
path, environ.get("HTTP_DEPTH", "0")))
path, environ.get("HTTP_DEPTH", "0"),
None, self._rights._user_groups))
# take root item for rights checking
item = next(items_iter, None)
if not item:

33
radicale/rights/from_file.py
View file

@ -49,33 +49,36 @@ class Rights(rights.BaseRights):
super().__init__(configuration)
self._filename = configuration.get("rights", "file")
self._log_rights_rule_doesnt_match_on_debug = configuration.get("logging", "rights_rule_doesnt_match_on_debug")
self._rights_config = configparser.ConfigParser()
try:
with open(self._filename, "r") as f:
self._rights_config.read_file(f)
logger.debug("Read rights file")
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:
user = user or ""
sane_path = pathutils.strip_path(path)
# Prevent "regex injection"
escaped_user = re.escape(user)
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
if not self._log_rights_rule_doesnt_match_on_debug:
logger.debug("logging of rules which doesn't match suppressed by config/option [logging] rights_rule_doesnt_match_on_debug")
for section in rights_config.sections():
group_match = False
for section in self._rights_config.sections():
group_match = None
user_match = None
try:
user_pattern = rights_config.get(section, "user")
collection_pattern = rights_config.get(section, "collection")
allowed_groups = rights_config.get(section, "groups", fallback="").split(",")
user_pattern = self._rights_config.get(section, "user", fallback="")
collection_pattern = self._rights_config.get(section, "collection")
allowed_groups = self._rights_config.get(section, "groups", fallback="").split(",")
try:
group_match = len(self._user_groups.intersection(allowed_groups)) > 0
except Exception:
pass
# Use empty format() for harmonized handling of curly braces
user_match = re.fullmatch(user_pattern.format(), user)
if user_pattern != "":
user_match = re.fullmatch(user_pattern.format(), user)
user_collection_match = user_match and re.fullmatch(
collection_pattern.format(
*(re.escape(s) for s in user_match.groups()),
@ -85,13 +88,13 @@ class Rights(rights.BaseRights):
raise RuntimeError("Error in section %r of rights file %r: "
"%s" % (section, self._filename, e)) from e
if user_match and user_collection_match:
permission = rights_config.get(section, "permissions")
permission = self._rights_config.get(section, "permissions")
logger.debug("Rule %r:%r matches %r:%r from section %r permission %r",
user, sane_path, user_pattern,
collection_pattern, section, permission)
return permission
if group_match and group_collection_match:
permission = rights_config.get(section, "permissions")
permission = self._rights_config.get(section, "permissions")
logger.debug("Rule %r:%r matches %r:%r from section %r permission %r by group membership",
user, sane_path, user_pattern,
collection_pattern, section, permission)

11
radicale/storage/__init__.py
View file

@ -26,8 +26,8 @@ Take a look at the class ``BaseCollection`` if you want to implement your own.
import json
import xml.etree.ElementTree as ET
from hashlib import sha256
from typing import (Iterable, Iterator, Mapping, Optional, Sequence, Set,
Tuple, Union, overload)
from typing import (Callable, ContextManager, Iterable, Iterator, Mapping,
Optional, Sequence, Set, Tuple, Union, overload)
import vobject
@ -282,8 +282,11 @@ class BaseStorage:
"""
self.configuration = configuration
def discover(self, path: str, depth: str = "0") -> Iterable[
"types.CollectionOrItem"]:
def discover(
self, path: str, depth: str = "0",
child_context_manager: Optional[
Callable[[str, Optional[str]], ContextManager[None]]] = None,
user_groups: Set[str] = set([])) -> Iterable["types.CollectionOrItem"]:
"""Discover a list of collections under the given ``path``.
``path`` is sanitized.

19
radicale/storage/multifilesystem/discover.py
View file

@ -16,9 +16,10 @@
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
import base64
import os
import posixpath
from typing import Callable, ContextManager, Iterator, Optional, cast
from typing import Callable, ContextManager, Iterator, Optional, Set, cast
from radicale import pathutils, types
from radicale.log import logger
@ -35,8 +36,10 @@ def _null_child_context_manager(path: str,
class StoragePartDiscover(StorageBase):
def discover(
self, path: str, depth: str = "0", child_context_manager: Optional[
Callable[[str, Optional[str]], ContextManager[None]]] = None
self, path: str, depth: str = "0",
child_context_manager: Optional[
Callable[[str, Optional[str]], ContextManager[None]]] = None,
user_groups: Set[str] = set([])
) -> Iterator[types.CollectionOrItem]:
# assert isinstance(self, multifilesystem.Storage)
if child_context_manager is None:
@ -102,3 +105,13 @@ class StoragePartDiscover(StorageBase):
with child_context_manager(sane_child_path, None):
yield self._collection_class(
cast(multifilesystem.Storage, self), child_path)
for group in user_groups:
href = base64.b64encode(group.encode('utf-8')).decode('ascii')
logger.debug(f"searching for group calendar {group} {href}")
sane_child_path = f"GROUPS/{href}"
if not os.path.isdir(pathutils.path_to_filesystem(folder, sane_child_path)):
continue
child_path = f"/GROUPS/{href}/"
with child_context_manager(sane_child_path, None):
yield self._collection_class(
cast(multifilesystem.Storage, self), child_path)