Merge pull request #2195 from pbiering/sharing-by-group
Some checks failed
Build and publish Docker image / build-and-push-image (push) Has been cancelled
Cleanup old nightly docker images / Cleanup old nightly docker images (push) Has been cancelled

Sharing-by-group/realm
This commit is contained in:
Peter Bieringer
2026-08-08 08:36:23 +03:00
committed by GitHub
30 changed files with 1328 additions and 141 deletions

View File

@@ -3,6 +3,11 @@
## 3.8.0.dev ## 3.8.0.dev
* Fix: storage/multifilesystem: depth:1 PROPFIND no longer re-runs the filesystem collision check (path_to_filesystem) for every item in a collection; this made listing large collections O(n^2) on file systems not detected as collision-free * Fix: storage/multifilesystem: depth:1 PROPFIND no longer re-runs the filesystem collision check (path_to_filesystem) for every item in a collection; this made listing large collections O(n^2) on file systems not detected as collision-free
* Improve: storage/multifilesystem: avoid redundant stat() calls per item in get/upload when use_mtime_and_size_for_item_cache is enabled * Improve: storage/multifilesystem: avoid redundant stat() calls per item in get/upload when use_mtime_and_size_for_item_cache is enabled
* Feature: [sharing] add sharing-by-group/realm
* Feature: [group] with type "htgroup", "none", "from_auth" (NEW)
* Extension: [auth] type "pam": set groups of user to be used later
* Adjustment: reject usernames starting or ending with "@" or having more than one "@"
* Adjustment: reject usernames containing ":"
## 3.7.8 ## 3.7.8
* Fix: time-range filter on a VTODO having DTSTART/DUE and also CREATED/COMPLETED used the CREATED->COMPLETED duration instead of the DTSTART->DUE one, so completed tasks were missing from (or wrongly returned by) calendar-query REPORT results * Fix: time-range filter on a VTODO having DTSTART/DUE and also CREATED/COMPLETED used the CREATED->COMPLETED duration instead of the DTSTART->DUE one, so completed tasks were missing from (or wrongly returned by) calendar-query REPORT results

View File

@@ -1026,6 +1026,8 @@ Available types are:
* `ldap` _(>= 3.3.0)_ * `ldap` _(>= 3.3.0)_
Use a LDAP or AD server to authenticate users by relaying credentials from clients and handle results. Use a LDAP or AD server to authenticate users by relaying credentials from clients and handle results.
User groups are supported. Requires group/type=`from_auth` _(>= 3.8.0)_.
* `dovecot` _(>= 3.3.1)_ * `dovecot` _(>= 3.3.1)_
Use a Dovecot server to authenticate users by relaying credentials from clients and handle results. Use a Dovecot server to authenticate users by relaying credentials from clients and handle results.
@@ -1038,7 +1040,9 @@ Available types are:
in combination with SSO support in reverse proxy (e.g. Apache+mod_auth_openidc). in combination with SSO support in reverse proxy (e.g. Apache+mod_auth_openidc).
* `pam` _(>= 3.5.0)_ * `pam` _(>= 3.5.0)_
Use local PAM to authenticate users by relaying credentials from client and handle result.. Use local PAM to authenticate users by relaying credentials from client and handle result.
User groups are supported _(>= 3.8.0)_
Default: `none` _(< 3.5.0)_ / `denyall` _(>= 3.5.0)_ Default: `none` _(< 3.5.0)_ / `denyall` _(>= 3.5.0)_
@@ -1261,6 +1265,8 @@ They also give you access to the group calendars, if those exist.
Default: (unset) Default: (unset)
Requires group lookup type set to `from_auth` _(>= 3.8.0)_
##### ldap_group_members_attribute ##### ldap_group_members_attribute
_(>= 3.5.6)_ _(>= 3.5.6)_
@@ -1494,6 +1500,44 @@ This setting forces decoding the username.
Default: `False` Default: `False`
#### [group]
_(>= 3.8.0)_
##### type
The method to lookup groups for username
Available types are:
* `none`
No groups lookup at all
* `from_auth`
Group lookup by authentication type (if supported)
* `htgroup`
Use an
[Apache htgroup file](https://httpd.apache.org/docs/2.4/mod/mod_authz_groupfile.html)
to store groups and their members
Default: `none`
##### htgroup_filename
_(>= 3.8.0)_
Path to the htgroup file.
Default: `/etc/radicale/groups`
##### htgroup_cache
_(>= 3.8.0)_
Enable caching of htgroup file based on size and mtime_ns
Default: `False`
#### [rights] #### [rights]
@@ -2308,6 +2352,8 @@ Default: `false`
* If `False` it can be explicitly granted by *share* permissions: `P` * If `False` it can be explicitly granted by *share* permissions: `P`
* If `True` it can be explicitly forbidden by *share* permissions: `p` * If `True` it can be explicitly forbidden by *share* permissions: `p`
share-by-group/realm: always forbidden (_>= 3.8.0_)
##### enforce_properties_overlay ##### enforce_properties_overlay
_(>= 3.7.0)_ _(>= 3.7.0)_
@@ -2319,6 +2365,8 @@ Default: `true`
* If `False` it can be explicitly enforced by *share* permissions: `E` * If `False` it can be explicitly enforced by *share* permissions: `E`
* If `True` it can be explicitly forbidden by *share* permissions: `e` * If `True` it can be explicitly forbidden by *share* permissions: `e`
share-by-group/realm: always forbidden (_>= 3.8.0_)
##### default_permissions_create_token ##### default_permissions_create_token
_(>= 3.7.0)_ _(>= 3.7.0)_

View File

@@ -8,6 +8,10 @@ With _3.7.0_ major extension was implemented
* added management API * added management API
* WebUI extension using the management API * WebUI extension using the management API
With _3.8.0_ sharing-by-* membership was implemented
* sharing-by-group
* sharing-by-realm
## Sharing Implementation ## Sharing Implementation
Implementation of sharing collections is done by using a database to lookup the URI and in case entry exists by mapping to target URI and replacing provided data on request and adjust if required data in response. Implementation of sharing collections is done by using a database to lookup the URI and in case entry exists by mapping to target URI and replacing provided data on request and adjust if required data in response.
@@ -32,17 +36,22 @@ Types of supported sharing configuration:
* `map`: map-based share (requires user authentication) * `map`: map-based share (requires user authentication)
* `PathOrToken`: token or "virtual" collection, has to be unique (PRIMARY KEY) * `PathOrToken`: token or "virtual" collection, has to be unique (PRIMARY KEY)
* `PathMapped`: target collection * `PathMapped`: target collection
* share-by-group/realm: has to start with placeholder `/{user}` (_>= 3.8.0_)
* `Conversion`: conversion method * `Conversion`: conversion method
* `Owner`: owner of the share * `Owner`: owner of the share
* `User`: user of the share * `User`: user (or group, _>= 3.8.0_) of the share
* share-by-group/realm: has to start with `:` or `@` (_>= 3.8.0_)
* `Permissions`: effective permission of the share * `Permissions`: effective permission of the share
* `EnabledByOwner`: control by owner * `EnabledByOwner`: control by owner
* `EnabledByUser`: control by user * `EnabledByUser`: control by user
* share-by-group/realm: always enabled (_>= 3.8.0_)
* `HiddenByOwner`: control by owner * `HiddenByOwner`: control by owner
* `HiddenByUser`: control by user * `HiddenByUser`: control by user
* share-by-group/realm: always disabled (_>= 3.8.0_)
* `TimestampCreated`: unixtime of creation * `TimestampCreated`: unixtime of creation
* `TimestampUpdated`: unixtime of last update * `TimestampUpdated`: unixtime of last update
* `Properties`: overlay properties (limited set whitelisted) * `Properties`: overlay properties (limited set whitelisted)
* share-by-group/realm: not supported (_>= 3.8.0_)
* `Actions`: specific configuration * `Actions`: specific configuration
`Enabled*`: _owner_ AND _user_ have to enable a share to become usable `Enabled*`: _owner_ AND _user_ have to enable a share to become usable

13
config
View File

@@ -211,6 +211,19 @@
#urldecode_username = False #urldecode_username = False
[group]
# Group lookup method
# Value: none | from_auth | htgroup
type = none
# Htgroup filename
#htgroup_filename = /etc/radicale/groups
# Enable caching of htgroup file based on size and mtime_ns
#htgroup_cache = False
[rights] [rights]
# Rights backend # Rights backend

View File

@@ -40,10 +40,11 @@ dependencies = [
[project.optional-dependencies] [project.optional-dependencies]
test = ["pytest>=7", "waitress", "bcrypt", "argon2-cffi"] test = ["pytest>=7", "waitress", "bcrypt", "argon2-cffi", "pam", "ldap3"]
bcrypt = ["bcrypt"] bcrypt = ["bcrypt"]
argon2 = ["argon2-cffi"] argon2 = ["argon2-cffi"]
ldap = ["ldap3"] ldap = ["ldap3"]
pam = ["pam"]
dev = ["flake8", "isort", "mypy", "pytest", "pytest-playwright", "html5validator"] dev = ["flake8", "isort", "mypy", "pytest", "pytest-playwright", "html5validator"]
[project.scripts] [project.scripts]

View File

@@ -74,6 +74,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
"""WSGI application.""" """WSGI application."""
_mask_passwords: bool _mask_passwords: bool
_urldecode_username: bool
_auth_delay: float _auth_delay: float
_delay_on_error: float _delay_on_error: float
_internal_server: bool _internal_server: bool
@@ -110,6 +111,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
if not os.access(os.environ['TEMP'], os.W_OK): if not os.access(os.environ['TEMP'], os.W_OK):
raise RuntimeError("TEMP found in environment, but not writable: %r" % os.environ['TEMP']) raise RuntimeError("TEMP found in environment, but not writable: %r" % os.environ['TEMP'])
self._mask_passwords = configuration.get("logging", "mask_passwords") self._mask_passwords = configuration.get("logging", "mask_passwords")
self._urldecode_username = configuration.get("auth", "urldecode_username")
self._delay_on_error = configuration.get("server", "delay_on_error") self._delay_on_error = configuration.get("server", "delay_on_error")
logger.info("delay_on_error set to: %.3f seconds", self._delay_on_error) logger.info("delay_on_error set to: %.3f seconds", self._delay_on_error)
self._max_content_length = configuration.get("server", "max_content_length") self._max_content_length = configuration.get("server", "max_content_length")
@@ -549,18 +551,11 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
self.configuration, environ, base64.b64decode( self.configuration, environ, base64.b64decode(
authorization.encode("ascii"))).split(":", 1) authorization.encode("ascii"))).split(":", 1)
if login and not app_base._check_user_format(self._storage, login, self._validate_user_value): if login and not app_base._check_user_format(self._storage, login, self._validate_user_value, self._urldecode_username):
info = "not compliant to %r" % self._validate_user_value info = "not compliant to %r" % self._validate_user_value
user = "" user = ""
else: else:
(user, info) = self._auth.login(login, password, context) or ("", "") if login else ("", "") (user, info) = self._auth.login(login, password, context) or ("", "") if login else ("", "")
if self.configuration.get("auth", "type") == "ldap":
try:
logger.debug("Groups received from LDAP: %r", ",".join(self._auth._ldap_groups))
self._rights._user_groups = self._auth._ldap_groups
except AttributeError:
pass
request_info: dict = { request_info: dict = {
"method": request_method, "method": request_method,
"login": login, # not 'user' in this step "login": login, # not 'user' in this step
@@ -602,6 +597,19 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
logger.info("Refused unsafe username: %r", user) logger.info("Refused unsafe username: %r", user)
user = "" user = ""
if user:
group_type = self.configuration.get("group", "type")
if group_type in ["htgroup"]:
self._rights._user_groups = self._group.groups(login) if login else set([])
elif group_type in ["from_auth"]:
auth_type = self.configuration.get("auth", "type")
if auth_type in ["ldap", "pam"]:
try:
logger.debug("Groups received from %r: %r", auth_type, ",".join(self._auth._groups))
self._rights._user_groups = self._auth._groups
except AttributeError:
pass
# Create principal collection # Create principal collection
if user: if user:
principal_path = "/%s/" % user principal_path = "/%s/" % user

View File

@@ -22,9 +22,10 @@ import sys
import unicodedata import unicodedata
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from typing import Optional, Union from typing import Optional, Union
from urllib.parse import unquote
from radicale import (auth, config, hook, httputils, log, pathutils, rights, from radicale import (auth, config, group, hook, httputils, log, pathutils,
sharing, storage, types, utils, web, xmlutils) rights, sharing, storage, types, utils, web, xmlutils)
from radicale.log import logger from radicale.log import logger
from radicale.rights import intersect from radicale.rights import intersect
@@ -84,8 +85,30 @@ def _check_format(self: storage.BaseStorage,
def _check_user_format(self: storage.BaseStorage, def _check_user_format(self: storage.BaseStorage,
user: str, user: str,
validation_type: str validation_type: str,
urldecode_username: bool,
enforceUser: bool = True,
) -> bool: ) -> bool:
logger.trace("_check_user_format investigate %r (urldecode_username=%r)", user, urldecode_username)
if urldecode_username:
user = unquote(user)
if (user.startswith(sharing.SHARING_SEPARATOR_GROUP) or user.startswith(sharing.SHARING_SEPARATOR_REALM)):
if enforceUser:
# group/realm identifiers
return False
else:
# strip 1st char
user = user[1:]
if enforceUser:
if user.count(sharing.SHARING_SEPARATOR_GROUP) > 0:
# not allowed (avoid injecting a group)
return False
elif user.count(sharing.SHARING_SEPARATOR_REALM) > 1:
# only allowed once
return False
if (user.endswith(sharing.SHARING_SEPARATOR_GROUP) or user.endswith(sharing.SHARING_SEPARATOR_REALM)):
# group/realm identifiers
return False
if validation_type == "strict": if validation_type == "strict":
return (re.search(USER_PATTERN_STRICT_RE, user) is not None) return (re.search(USER_PATTERN_STRICT_RE, user) is not None)
else: else:
@@ -116,6 +139,7 @@ class ApplicationBase:
configuration: config.Configuration configuration: config.Configuration
_auth: auth.BaseAuth _auth: auth.BaseAuth
_group: group.BaseGroup
_storage: storage.BaseStorage _storage: storage.BaseStorage
_rights: rights.BaseRights _rights: rights.BaseRights
_web: web.BaseWeb _web: web.BaseWeb
@@ -133,6 +157,7 @@ class ApplicationBase:
def __init__(self, configuration: config.Configuration) -> None: def __init__(self, configuration: config.Configuration) -> None:
self.configuration = configuration self.configuration = configuration
self._auth = auth.load(configuration) self._auth = auth.load(configuration)
self._group = group.load(configuration)
self._storage = storage.load(configuration) self._storage = storage.load(configuration)
self._rights = rights.load(configuration) self._rights = rights.load(configuration)
self._web = web.load(configuration) self._web = web.load(configuration)

View File

@@ -23,7 +23,7 @@ from http import client
from typing import Optional, Union from typing import Optional, Union
from urllib.parse import quote from urllib.parse import quote
from radicale import httputils, storage, types, xmlutils from radicale import httputils, sharing, storage, types, xmlutils
from radicale.app.base import Access, ApplicationBase from radicale.app.base import Access, ApplicationBase
from radicale.hook import HookNotificationItem, HookNotificationItemTypes from radicale.hook import HookNotificationItem, HookNotificationItemTypes
from radicale.log import logger from radicale.log import logger
@@ -71,6 +71,10 @@ class ApplicationPartDelete(ApplicationBase):
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']

View File

@@ -23,7 +23,7 @@ from http import client
from typing import Union from typing import Union
from urllib.parse import quote from urllib.parse import quote
from radicale import httputils, pathutils, storage, types, xmlutils from radicale import httputils, pathutils, sharing, storage, types, xmlutils
from radicale.app.base import Access, ApplicationBase from radicale.app.base import Access, ApplicationBase
from radicale.log import logger from radicale.log import logger
@@ -89,7 +89,10 @@ class ApplicationPartGet(ApplicationBase):
share = None share = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']

View File

@@ -24,7 +24,7 @@ import re
from http import client from http import client
from urllib.parse import unquote, urlparse from urllib.parse import unquote, urlparse
from radicale import httputils, pathutils, storage, types from radicale import httputils, pathutils, sharing, storage, types
from radicale.app import base as app_base from radicale.app import base as app_base
from radicale.app.base import Access, ApplicationBase from radicale.app.base import Access, ApplicationBase
from radicale.log import logger from radicale.log import logger
@@ -73,7 +73,10 @@ class ApplicationPartMove(ApplicationBase):
permissions_filter = None permissions_filter = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']
@@ -93,7 +96,10 @@ class ApplicationPartMove(ApplicationBase):
to_path = to_path[len(base_prefix):] to_path = to_path[len(base_prefix):]
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(to_path, to_user) to_user_lookup = to_user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
to_user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(to_path, to_user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
to_path = share['PathMapped'] to_path = share['PathMapped']

View File

@@ -27,8 +27,8 @@ from http import client
from typing import (Dict, Iterable, Iterator, List, Optional, Sequence, Tuple, from typing import (Dict, Iterable, Iterator, List, Optional, Sequence, Tuple,
Union) Union)
from radicale import (httputils, pathutils, rights, storage, types, utils, from radicale import (httputils, pathutils, rights, sharing, storage, types,
xmlutils) utils, xmlutils)
from radicale.app.base import Access, ApplicationBase from radicale.app.base import Access, ApplicationBase
from radicale.log import logger from radicale.log import logger
@@ -588,7 +588,10 @@ class ApplicationPartPropfind(ApplicationBase):
allowed_items: list = [] allowed_items: list = []
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']
@@ -639,7 +642,10 @@ class ApplicationPartPropfind(ApplicationBase):
if http_depth == "1": if http_depth == "1":
logger.trace("PROPFIND: get shared collections") logger.trace("PROPFIND: get shared collections")
# check for shared collections related to user, Enabled and not Hidden # check for shared collections related to user, Enabled and not Hidden
collections_share_list = self._sharing.sharing_collection_list(User=user, Enabled=True, Hidden=False) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
collections_share_list = self._sharing.sharing_collection_list(User=user_lookup, Enabled=True, Hidden=False)
if collections_share_list: if collections_share_list:
for share in collections_share_list: for share in collections_share_list:
c_share = share['PathOrToken'] c_share = share['PathOrToken']

View File

@@ -107,7 +107,10 @@ class ApplicationPartProppatch(ApplicationBase):
path_orig = path path_orig = path
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']

View File

@@ -33,8 +33,8 @@ from typing import Iterator, List, Mapping, MutableMapping, Optional, Tuple
import vobject import vobject
import radicale.item as radicale_item import radicale.item as radicale_item
from radicale import (httputils, pathutils, rights, storage, types, utils, from radicale import (httputils, pathutils, rights, sharing, storage, types,
xmlutils) utils, xmlutils)
from radicale.app.base import Access, ApplicationBase from radicale.app.base import Access, ApplicationBase
from radicale.hook import HookNotificationItem, HookNotificationItemTypes from radicale.hook import HookNotificationItem, HookNotificationItemTypes
from radicale.log import logger from radicale.log import logger
@@ -188,7 +188,10 @@ class ApplicationPartPut(ApplicationBase):
permissions_filter = None permissions_filter = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']

View File

@@ -38,7 +38,7 @@ import vobject.base
from vobject.base import ContentLine from vobject.base import ContentLine
import radicale.item as radicale_item import radicale.item as radicale_item
from radicale import httputils, pathutils, storage, types, xmlutils from radicale import httputils, pathutils, sharing, storage, types, xmlutils
from radicale.app.base import Access, ApplicationBase from radicale.app.base import Access, ApplicationBase
from radicale.item import filter as radicale_filter from radicale.item import filter as radicale_filter
from radicale.log import logger from radicale.log import logger
@@ -864,7 +864,10 @@ class ApplicationPartReport(ApplicationBase):
share = None share = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user) user_lookup = user
if self._rights._user_groups is not None and len(self._rights._user_groups) > 0:
user_lookup += sharing.SHARING_SEPARATOR_GROUP + ','.join(self._rights._user_groups)
share = self._sharing.sharing_collection_resolver(path, user_lookup)
if share: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = share['PathMapped'] path = share['PathMapped']

View File

@@ -106,7 +106,7 @@ class AuthContext:
class BaseAuth: class BaseAuth:
_ldap_groups: Set[str] = set([]) _groups: Set[str] = set([])
_urldecode_username: bool _urldecode_username: bool
_lc_username: bool _lc_username: bool
_uc_username: bool _uc_username: bool

View File

@@ -381,8 +381,8 @@ class Auth(auth.BaseAuth):
tmp.append(rdns[0][1]) tmp.append(rdns[0][1])
except Exception: except Exception:
tmp.append(g) tmp.append(g)
self._ldap_groups = set(tmp) self._groups = set(tmp)
logger.debug("_login3 LDAP groups of user: %s", ",".join(self._ldap_groups)) logger.debug("_login3 LDAP groups of user: %s", ",".join(self._groups))
if self._ldap_user_attr: if self._ldap_user_attr:
if user_entry['attributes'][self._ldap_user_attr]: if user_entry['attributes'][self._ldap_user_attr]:

View File

@@ -97,6 +97,11 @@ class Auth(auth.BaseAuth):
else: else:
logger.debug("PAM user %r belongs to the required group: %r" % (login, self._group_membership)) logger.debug("PAM user %r belongs to the required group: %r" % (login, self._group_membership))
# add groups
members.append(primary_group)
self._groups = set(members)
logger.debug("PAM groups of user: %s", ",".join(self._groups))
# Check the password # Check the password
if self.pam_authenticate(login, password, service=self._service): if self.pam_authenticate(login, password, service=self._service):
return login return login

View File

@@ -39,8 +39,8 @@ from configparser import RawConfigParser
from typing import (Any, Callable, ClassVar, Iterable, List, Optional, from typing import (Any, Callable, ClassVar, Iterable, List, Optional,
Sequence, Tuple, TypeVar, Union) Sequence, Tuple, TypeVar, Union)
from radicale import (auth, hook, log, rights, sharing, storage, types, utils, from radicale import (auth, group, hook, log, rights, sharing, storage, types,
web) utils, web)
from radicale.hook import email from radicale.hook import email
from radicale.item import check_and_sanitize_props from radicale.item import check_and_sanitize_props
@@ -484,6 +484,21 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
"value": "False", "value": "False",
"help": "url-decode the username, set to True when clients send url-encoded email address as username", "help": "url-decode the username, set to True when clients send url-encoded email address as username",
"type": bool})])), "type": bool})])),
("group", OrderedDict([
("type", {
"value": "none",
"help": "group lookup method (" + "|".join(group.INTERNAL_TYPES) + ")",
"type": str_or_callable,
"internal": group.INTERNAL_TYPES}),
("htgroup_filename", {
"value": "/etc/radicale/groups",
"help": "htpgroup filename",
"type": filepath}),
("htgroup_cache", {
"value": "False",
"help": "enable caching of htgroup file",
"type": bool}),
])),
("rights", OrderedDict([ ("rights", OrderedDict([
("type", { ("type", {
"value": "owner_only", "value": "owner_only",

View File

@@ -0,0 +1,72 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
"""
Group membership module.
Enrich user with group membership
Take a look at the class ``BaseGroup`` if you want to implement your own.
"""
from typing import Sequence, Set, final
from radicale import config, utils
from radicale.log import logger
INTERNAL_TYPES: Sequence[str] = ("none",
"from_auth",
"htgroup",
)
def load(configuration: "config.Configuration") -> "BaseGroup":
"""Load the group module chosen in configuration."""
_type = configuration.get("group", "type")
if _type == "none":
logger.info("No user groups lookup method is selected")
else:
logger.info("User groups lookup method: %r", _type)
return utils.load_plugin(INTERNAL_TYPES, "group", "Group", BaseGroup,
configuration)
class BaseGroup:
def __init__(self, configuration: "config.Configuration") -> None:
"""Initialize BaseGroup.
``configuration`` see ``radicale.config`` module.
The ``configuration`` must not change during the lifetime of
this object, it is kept as an internal reference.
"""
self.configuration = configuration
self._type = configuration.get("group", "type")
def _groups(self, login: str) -> Set[str]:
"""Retrieve set of groups of a user
``login`` the login name
"""
raise NotImplementedError
@final
def groups(self, login: str) -> Set[str]:
return self._groups(login)

View File

@@ -0,0 +1,35 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
"""
A dummy backend that returns no group but check whether authentication type supports it.
"""
from typing import Set
from radicale import config, group
class Group(group.BaseGroup):
def __init__(self, configuration: config.Configuration) -> None:
super().__init__(configuration)
auth_type = configuration.get("auth", "type")
if auth_type not in ["ldap", "pam"]:
raise RuntimeError("group-type 'auth_type' is not supported by auth/type %r" % auth_type)
def _groups(self, login: str) -> Set[str]:
return set([])

182
radicale/group/htgroup.py Normal file
View File

@@ -0,0 +1,182 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
"""
Backend that retrieves groups of a user from htgroups file.
Apache's htgroup format (https://httpd.apache.org/docs/2.4/mod/mod_authz_groupfile.html)
"""
import os
import threading
import time
from typing import Set, Tuple
from radicale import config, group, logger
class Group(group.BaseGroup):
_filename: str
_encoding: str
_htgroup_by_member: dict[str, Set] # member -> groups (set)
_htgroup_mtime_ns: int
_htgroup_size: int
_htgroup_ok: bool
_htgroup_not_ok_time: float
_htgroup_not_ok_reminder_seconds: int
_htgroup_cache: bool
_lock: threading.Lock
def __init__(self, configuration: config.Configuration) -> None:
super().__init__(configuration)
self._filename = configuration.get("group", "htgroup_filename")
logger.info("group htgroup file: %r", self._filename)
self._encoding = configuration.get("encoding", "stock")
logger.info("group htgroup file encoding: %r", self._encoding)
self._htgroup_cache = configuration.get("group", "htgroup_cache")
logger.info("group htgroup cache: %s", self._htgroup_cache)
self._htgroup_ok = False
self._htgroup_not_ok_reminder_seconds = 60 # currently hardcoded
(self._htgroup_ok, self._htgroup_by_member, self._htgroup_size, self._htgroup_mtime_ns) = self._read_htgroup(True, False)
self._lock = threading.Lock()
def _read_htgroup(self, init: bool, suppress: bool) -> Tuple[bool, dict, int, int]:
"""Read htgroup file
init == True: stop on error
init == False: warn/skip on error and set mark to log reminder every interval
suppress == True: suppress warnings, change info to debug (used in non-caching mode)
suppress == False: do not suppress warnings (used in caching mode)
"""
htgroup_ok = True
if (init is True) or (suppress is True):
info = "Read"
else:
info = "Re-read"
if suppress is False:
logger.info("%s content of htgroup file start: %r", info, self._filename)
else:
logger.debug("%s content of htgroup file start: %r", info, self._filename)
htgroup: dict[str, str] = dict()
htgroup_by_member: dict[str, Set[str]] = dict()
entries = 0
duplicates = 0
errors = 0
try:
with open(self._filename, encoding=self._encoding) as f:
line_num = 0
for line in f:
line_num += 1
line = line.rstrip("\n")
if line.lstrip() and not line.lstrip().startswith("#"):
try:
group, members = line.split(":", maxsplit=1)
skip = False
if group == "":
if init is True:
raise ValueError("htgroup file contains problematic line not matching <group>:<members> in line: %d" % line_num)
else:
errors += 1
logger.warning("htgroup file contains problematic line not matching <group>:<members> in line: %d (ignored)", line_num)
htgroup_ok = False
skip = True
else:
if htgroup.get(group):
duplicates += 1
if init is True:
raise ValueError("htgroup file contains duplicate group: '%s'", group, line_num)
else:
logger.warning("htgroup file contains duplicate group: '%s' (line: %d / ignored)", group, line_num)
htgroup_ok = False
skip = True
if skip is False:
htgroup[group] = members
entries += 1
except ValueError as e:
if init is True:
raise RuntimeError("Invalid htgroup file %r: %s" % (self._filename, e)) from e
except OSError as e:
if init is True:
raise RuntimeError("Failed to load htgroup file %r: %s" % (self._filename, e)) from e
else:
logger.warning("Failed to load htgroup file on re-read: %r" % self._filename)
htgroup_ok = False
htgroup_size = os.stat(self._filename).st_size
htgroup_mtime_ns = os.stat(self._filename).st_mtime_ns
if suppress is False:
logger.info("%s content of htgroup file done: %r (entries: %d, duplicates: %d, errors: %d)", info, self._filename, entries, duplicates, errors)
else:
logger.debug("%s content of htgroup file done: %r (entries: %d, duplicates: %d, errors: %d)", info, self._filename, entries, duplicates, errors)
if htgroup_ok is True:
self._htgroup_not_ok_time = 0
else:
self._htgroup_not_ok_time = time.time()
# convert mapping
for group in htgroup:
for member in htgroup[group].split(' '):
if member not in htgroup_by_member:
htgroup_by_member[member] = set([group])
else:
htgroup_by_member[member].add(group)
return (htgroup_ok, htgroup_by_member, htgroup_size, htgroup_mtime_ns)
def _groups(self, login: str) -> Set[str]:
"""Get list of groups of login
Optional: the content of the file is cached and live updates will be detected by
comparing mtime_ns and size
"""
logger.trace("Group memberships (htgroup) lookup for user %r", login)
group_ok = False
groups: Set[str]
if self._htgroup_cache is True:
# check and re-read file if required
with self._lock:
htgroup_size = os.stat(self._filename).st_size
htgroup_mtime_ns = os.stat(self._filename).st_mtime_ns
if (htgroup_size != self._htgroup_size) or (htgroup_mtime_ns != self._htgroup_mtime_ns):
(self._htgroup_ok, self._htgroup, self._htgroup_size, self._htgroup_mtime_ns) = self._read_htgroup(False, False)
self._htgroup_not_ok_time = 0
# log reminder of problemantic file every interval
current_time = time.time()
if (self._htgroup_ok is False):
if (self._htgroup_not_ok_time > 0):
if (current_time - self._htgroup_not_ok_time) > self._htgroup_not_ok_reminder_seconds:
logger.warning("htgroup file still contains issues (REMINDER, check warnings in the past): %r" % self._filename)
self._htgroup_not_ok_time = current_time
else:
self._htgroup_not_ok_time = current_time
if self._htgroup_by_member.get(login):
groups = self._htgroup_by_member[login]
group_ok = True
else:
# read file on every request
(htgroup_ok, htgroup_by_member, htgroup_size, htgroup_mtime_ns) = self._read_htgroup(False, True)
if htgroup_by_member.get(login):
groups = htgroup_by_member[login]
group_ok = True
if group_ok is True:
logger.debug("Group memberships (htgroup) for user %r: %r", login, groups)
return groups
else:
logger.debug("Group memberships (htgroup) for user %r not found", login)
return set([])

30
radicale/group/none.py Normal file
View File

@@ -0,0 +1,30 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
"""
A dummy backend that returns no group.
"""
from typing import Set
from radicale import group
class Group(group.BaseGroup):
def _groups(self, login: str) -> Set[str]:
return set([])

View File

@@ -69,6 +69,32 @@ def intersect(a: str, b: str) -> str:
return "".join(set(a).intersection(set(b))) return "".join(set(a).intersection(set(b)))
def remove(a: str, b: str) -> str:
"""Remove rights from a defined in b
Returns all rights of ``a`` not listed in ``b``.
"""
result = set(a)
for entry in set(b):
if entry in a:
result.remove(entry)
return "".join(result)
def add(a: str, b: str) -> str:
"""Add rights to a defined in b
Returns all rights of ``a`` and ``b``.
"""
result = set(a)
for entry in set(b):
if entry not in a:
result.add(entry)
return "".join(result)
class BaseRights: class BaseRights:
_user_groups: Set[str] = set([]) _user_groups: Set[str] = set([])

View File

@@ -131,6 +131,9 @@ SHARING_BDAY_DESCRIPTION_TEMPLATE_DEFAULT: str = "BDAY={year}-{month}-{day}"
SHARING_BDAY_CATEGORIES_DEFAULT: str = 'Birthday' SHARING_BDAY_CATEGORIES_DEFAULT: str = 'Birthday'
SHARING_ACTIONS_DELETE_VALUE: str = '#DEL#' SHARING_ACTIONS_DELETE_VALUE: str = '#DEL#'
SHARING_SEPARATOR_REALM: str = '@'
SHARING_SEPARATOR_GROUP: str = ':'
def check_bday_max_age(data: Any) -> int: def check_bday_max_age(data: Any) -> int:
value = int(data) value = int(data)
@@ -915,7 +918,7 @@ class BaseSharing:
elif not request_data[key].endswith("/"): elif not request_data[key].endswith("/"):
return httputils.bad_request("PathMapped not ending with /") return httputils.bad_request("PathMapped not ending with /")
elif key == "User": elif key == "User":
if not app_base._check_user_format(self._storage, request_data[key], self._validate_user_value): if not app_base._check_user_format(self._storage, request_data[key], self._validate_user_value, enforceUser=False, urldecode_username=False):
logger.warning("%s: invalid %r: %r (not compliant to %r)", api_info, key, request_data[key], self._validate_user_value) logger.warning("%s: invalid %r: %r (not compliant to %r)", api_info, key, request_data[key], self._validate_user_value)
return httputils.bad_request("Invalid value for User") return httputils.bad_request("Invalid value for User")
@@ -1090,10 +1093,9 @@ class BaseSharing:
Permissions = str(Permissions) Permissions = str(Permissions)
if Conversion == "bday": if Conversion == "bday":
# bday is read-only and not supporting "Ee" # bday is read-only and not supporting "Ee"
for permission in Permissions: if rights.intersect(Permissions, "Eew"):
if permission not in "rPp":
logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for Conversion=%r", PathMapped, Permissions, Conversion) logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for Conversion=%r", PathMapped, Permissions, Conversion)
return httputils.bad_request("Permissions are not supported for conversion") return httputils.bad_request("Permissions are not supported for conversion: %r" % Permissions)
if Enabled is None: if Enabled is None:
Enabled = False # security by default Enabled = False # security by default
@@ -1183,6 +1185,14 @@ class BaseSharing:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'M')", PathMapped, user) logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'M')", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if User.startswith(SHARING_SEPARATOR_GROUP) or User.startswith(SHARING_SEPARATOR_REALM):
if PathOrToken.startswith("/{user}/"):
# placeholder exists
pass
else:
logger.warning(api_info + ": PathOrToken=%r has to start with placeholder for 'user' using group User=%r", PathOrToken, User)
return httputils.NOT_ALLOWED
else:
access = Access(self._rights, User, PathOrToken) access = Access(self._rights, User, PathOrToken)
if not access.check("r"): if not access.check("r"):
logger.warning(api_info + ": access to PathOrToken=%r not allowed for User=%r", PathOrToken, User) logger.warning(api_info + ": access to PathOrToken=%r not allowed for User=%r", PathOrToken, User)
@@ -1197,6 +1207,15 @@ class BaseSharing:
logger.warning(api_info + ": PathOrToken=%r already exists as real collection for User=%r", PathOrToken, User) logger.warning(api_info + ": PathOrToken=%r already exists as real collection for User=%r", PathOrToken, User)
return httputils.CONFLICT return httputils.CONFLICT
if User.startswith(SHARING_SEPARATOR_GROUP) or User.startswith(SHARING_SEPARATOR_REALM):
# enforce user toggles for groups
HiddenByUser = False
EnabledByUser = True
if rights.intersect(Permissions, "EP"):
logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for share-by-group/realm", PathMapped, Permissions)
return httputils.bad_request("Permissions are not supported for conversion: %r" % Permissions)
Permissions = rights.add(Permissions, "ep") # enforce permissions for group
logger.trace("" + api_info + ": %r (Permissions=%r PathOrToken=%r Owner=%r User=%r)", PathMapped, Permissions, PathOrToken, user, User) logger.trace("" + api_info + ": %r (Permissions=%r PathOrToken=%r Owner=%r User=%r)", PathMapped, Permissions, PathOrToken, user, User)
result = self.database_create_sharing( result = self.database_create_sharing(
@@ -1308,16 +1327,23 @@ class BaseSharing:
Permissions = str(Permissions) Permissions = str(Permissions)
if share['Conversion'] == "bday": if share['Conversion'] == "bday":
# bday is read-only and not supporting "Ee" # bday is read-only and not supporting "Ee"
for permission in Permissions: if rights.intersect(Permissions, "Eew"):
if permission not in "rPp":
logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for Conversion=%r", PathMapped, Permissions, Conversion) logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for Conversion=%r", PathMapped, Permissions, Conversion)
return httputils.bad_request("Permissions are not supported for conversion") return httputils.bad_request("Permissions are not supported for conversion: %r" % Permissions)
if Conversion is not None and share['Conversion'] is not None: if Conversion is not None and share['Conversion'] is not None:
if Conversion != share['Conversion']: if Conversion != share['Conversion']:
logger.warning(api_info + ": PathMapped=%r change of Conversion %r -> %r is not supported", PathMapped, share['Conversion'], Conversion) logger.warning(api_info + ": PathMapped=%r change of Conversion %r -> %r is not supported", PathMapped, share['Conversion'], Conversion)
return httputils.bad_request("Change of conversion is not supported") return httputils.bad_request("Change of conversion is not supported")
if (User is not None and (User.startswith(SHARING_SEPARATOR_GROUP) or User.startswith(SHARING_SEPARATOR_REALM))) or (share['User'].startswith(SHARING_SEPARATOR_GROUP) or share['User'].startswith(SHARING_SEPARATOR_REALM)):
# enforce user permissions for groups
if Permissions is not None:
if rights.intersect(Permissions, "EP"):
logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for share-by-group/realm", PathMapped, Permissions)
return httputils.bad_request("Permissions are not supported for share-by-group/realm: %r" % Permissions)
Permissions = rights.add(Permissions, "ep") # enforce permissions for group
if user == share['Owner']: if user == share['Owner']:
if PathMapped is not None: if PathMapped is not None:
# check access Permissions # check access Permissions

View File

@@ -95,37 +95,21 @@ class Sharing(sharing.BaseSharing):
OnlyEnabled: bool = True, OnlyEnabled: bool = True,
User: Union[str, None] = None) -> Union[dict, None]: User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """ """ retrieve sharing target and attributes by map """
# Lookup logger.trace("sharing/%s/get: PathOrToken=%r User=%r OnlyEnabled=%s", ShareType, PathOrToken, User, OnlyEnabled)
logger.trace("sharing: lookup ShareType=%r PathOrToken=%r User=%r OnlyEnabled=%s)", ShareType, PathOrToken, User, OnlyEnabled)
index = 0
found = False found = False
for row in self._sharing_cache: for row in self.database_list_sharing(ShareType=ShareType, PathOrToken=PathOrToken, User=User):
if index == 0: # run through prefiltered list
# skip fieldnames logger.trace("sharing/get/check: %r", row)
pass if OnlyEnabled is True and row['EnabledByOwner'] is False:
else: continue
logger.trace("sharing: check row: %r", row) elif OnlyEnabled is True and row['EnabledByUser'] is False:
if row['ShareType'] != ShareType: continue
pass
elif row['PathOrToken'] != PathOrToken:
pass
elif User is not None and row['User'] != User:
pass
elif OnlyEnabled is True and row['EnabledByOwner'] is not True:
pass
elif OnlyEnabled is True and row['EnabledByUser'] is not True:
pass
else: else:
found = True found = True
break break
index += 1
if found: if found:
PathMapped = row['PathMapped']
Owner = row['Owner']
UserShare = row['User']
Permissions = row['Permissions']
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser']) Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
Properties: Union[dict, None] = None Properties: Union[dict, None] = None
Conversion: Union[str, None] = None Conversion: Union[str, None] = None
@@ -140,13 +124,13 @@ class Sharing(sharing.BaseSharing):
"mapped": True, "mapped": True,
"ShareType": ShareType, "ShareType": ShareType,
"PathOrToken": PathOrToken, "PathOrToken": PathOrToken,
"PathMapped": PathMapped, "PathMapped": row['PathMapped'],
"Owner": Owner, "Owner": row['Owner'],
"User": UserShare, "User": row['User'],
"Hidden": Hidden, "Hidden": Hidden,
"EnabledByOwner": row['EnabledByOwner'], "EnabledByOwner": row['EnabledByOwner'],
"EnabledByUser": row['EnabledByUser'], "EnabledByUser": row['EnabledByUser'],
"Permissions": Permissions, "Permissions": row['Permissions'],
"Properties": Properties, "Properties": Properties,
"Conversion": Conversion, "Conversion": Conversion,
"Actions": Actions, "Actions": Actions,
@@ -174,39 +158,79 @@ class Sharing(sharing.BaseSharing):
logger.trace("sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s Conversion=%r", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser, Conversion) logger.trace("sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s Conversion=%r", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser, Conversion)
for row in self._sharing_cache: for row in self._sharing_cache:
if index == 0:
# skip fieldnames
pass
else:
logger.trace("sharing/list/row: test: %r", row)
if ShareType is not None and row['ShareType'] != ShareType:
logger.trace("sharing/list/row: skip by ShareType")
pass
elif OwnerOrUser is not None and (row['Owner'] != OwnerOrUser and row['User'] != OwnerOrUser):
pass
elif User is not None and row['User'] != User:
logger.trace("sharing/list/row: skip by User")
pass
elif PathOrToken is not None and row['PathOrToken'] != PathOrToken:
logger.trace("sharing/list/row: skip by PathOrToken")
pass
elif PathMapped is not None and row['PathMapped'] != PathMapped:
logger.trace("sharing/list/row: skip by PathMapped")
pass
elif EnabledByOwner is not None and row['EnabledByOwner'] != EnabledByOwner:
pass
elif EnabledByUser is not None and row['EnabledByUser'] != EnabledByUser:
pass
elif HiddenByOwner is not None and row['HiddenByOwner'] != HiddenByOwner:
pass
elif HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser:
pass
elif Conversion is not None and row['Conversion'] != Conversion:
pass
else:
logger.trace("sharing/list/row: add : %r", row)
result.append(row)
index += 1 index += 1
if index == 1:
# skip fieldnames
continue
logger.trace("sharing/list/row: test: %r", row)
if ShareType is not None and row['ShareType'] != ShareType:
continue
if Conversion is not None and row['Conversion'] != Conversion:
continue
if EnabledByOwner is not None and row['EnabledByOwner'] != EnabledByOwner:
continue
if EnabledByUser is not None and row['EnabledByUser'] != EnabledByUser:
continue
if HiddenByOwner is not None and row['HiddenByOwner'] != HiddenByOwner:
continue
if HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser:
continue
if PathMapped is not None and row['PathMapped'] != PathMapped:
continue
if OwnerOrUser is not None:
if User is not None and OwnerOrUser == User:
pass # will be checked below
elif (row['Owner'] != OwnerOrUser) and (row['User'] != OwnerOrUser):
continue
group_check = False
if row['User'].startswith(sharing.SHARING_SEPARATOR_GROUP) or row['User'].startswith(sharing.SHARING_SEPARATOR_REALM):
group_check = True
if User is not None:
if row['User'].startswith(sharing.SHARING_SEPARATOR_REALM):
if not User.endswith(row['User']):
continue
else:
pass
elif row['User'].startswith(sharing.SHARING_SEPARATOR_GROUP):
if sharing.SHARING_SEPARATOR_GROUP not in User:
continue # user has no group
user_without_group = User.split(sharing.SHARING_SEPARATOR_GROUP)[0]
groups_of_user = User.split(sharing.SHARING_SEPARATOR_GROUP)[1].split(',')
Groups = row['User'].removeprefix(sharing.SHARING_SEPARATOR_GROUP).split(',')
logger.trace("sharing/list/check/groups: groups_of_user=%r Groups=%r", groups_of_user, Groups)
found = False
for group in groups_of_user:
if group in Groups:
found = True
break
if found:
pass
else:
continue
elif row['User'] == User:
pass
else:
continue
row_copy = row.copy()
if group_check and User is not None:
if row['User'].startswith(sharing.SHARING_SEPARATOR_GROUP):
user_without_group = User.split(sharing.SHARING_SEPARATOR_GROUP)[0]
else:
user_without_group = User
row_copy['PathOrToken'] = row['PathOrToken'].replace("{user}", user_without_group) # replace placeholder
row_copy['User'] = user_without_group # replace with real user
if PathOrToken is not None and row_copy['PathOrToken'] != PathOrToken:
continue
logger.trace("sharing/list/row: add : %r", row_copy)
result.append(row_copy)
return result return result
def database_create_sharing(self, def database_create_sharing(self,

View File

@@ -91,12 +91,36 @@ class Sharing(sharing.BaseSharing):
OnlyEnabled: bool = True, OnlyEnabled: bool = True,
User: Union[str, None] = None) -> Union[dict, None]: User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """ """ retrieve sharing target and attributes by map """
# Lookup
logger.trace("sharing/%s/get: PathOrToken=%r User=%r)", ShareType, PathOrToken, User)
sharing_config_file = os.path.join(self._sharing_database_path_ShareType[ShareType], self._encode_path(PathOrToken)) sharing_config_file = os.path.join(self._sharing_database_path_ShareType[ShareType], self._encode_path(PathOrToken))
logger.trace("sharing/%s/get: PathOrToken=%r User=%r OnlyEnabled=%s -> config=%r)", ShareType, PathOrToken, User, OnlyEnabled, sharing_config_file)
if not os.path.isfile(sharing_config_file): if not os.path.isfile(sharing_config_file):
if ShareType != "map" or User is None:
return None
else:
# check by group
logger.trace("sharing/%s/get: no direct share found, run through filtered list")
for row in self.database_list_sharing(ShareType=ShareType, PathOrToken=PathOrToken, User=User):
if OnlyEnabled is True and row['EnabledByOwner'] is False:
continue
if OnlyEnabled is True and row['EnabledByUser'] is False:
continue
return {
"mapped": True,
"ShareType": ShareType,
"PathOrToken": row['PathOrToken'],
"PathMapped": row['PathMapped'],
"Owner": row['Owner'],
"User": row['User'],
"Hidden": row['HiddenByOwner'],
"EnabledByOwner": row['EnabledByOwner'],
"EnabledByUser": row['EnabledByUser'],
"Permissions": row['Permissions'],
"Properties": row['Properties'],
"Conversion": row['Conversion'],
"Actions": row['Actions'],
}
return None return None
# read content # read content
@@ -187,33 +211,77 @@ class Sharing(sharing.BaseSharing):
continue continue
logger.trace("sharing/list/row: test: %r", row) logger.trace("sharing/list/row: test: %r", row)
if ShareType is not None and row['ShareType'] != ShareType: if ShareType is not None and row['ShareType'] != ShareType:
logger.trace("sharing/list/row: skip by ShareType") continue
pass if Conversion is not None and row['Conversion'] != Conversion:
elif OwnerOrUser is not None and (row['Owner'] != OwnerOrUser and row['User'] != OwnerOrUser): continue
pass if EnabledByOwner is not None and row['EnabledByOwner'] != EnabledByOwner:
elif User is not None and row['User'] != User: continue
logger.trace("sharing/list/row: skip by User") if EnabledByUser is not None and row['EnabledByUser'] != EnabledByUser:
pass continue
elif PathOrToken is not None and row['PathOrToken'] != PathOrToken: if HiddenByOwner is not None and row['HiddenByOwner'] != HiddenByOwner:
logger.trace("sharing/list/row: skip by PathOrToken") continue
pass if HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser:
elif PathMapped is not None and row['PathMapped'] != PathMapped: continue
logger.trace("sharing/list/row: skip by PathMapped") if PathMapped is not None and row['PathMapped'] != PathMapped:
pass continue
elif EnabledByOwner is not None and row['EnabledByOwner'] != EnabledByOwner: if OwnerOrUser is not None:
pass if User is not None and OwnerOrUser == User:
elif EnabledByUser is not None and row['EnabledByUser'] != EnabledByUser: pass # will be checked below
pass elif (row['Owner'] != OwnerOrUser) and (row['User'] != OwnerOrUser):
elif HiddenByOwner is not None and row['HiddenByOwner'] != HiddenByOwner: continue
pass
elif HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser: group_check = False
pass if row['User'].startswith(sharing.SHARING_SEPARATOR_GROUP) or row['User'].startswith(sharing.SHARING_SEPARATOR_REALM):
elif Conversion is not None and row['Conversion'] != Conversion: group_check = True
if User is not None:
if row['User'].startswith(sharing.SHARING_SEPARATOR_REALM):
if not User.endswith(row['User']):
continue
elif row['User'].startswith(sharing.SHARING_SEPARATOR_GROUP):
if sharing.SHARING_SEPARATOR_GROUP not in User:
continue # user has no group
user_without_group = User.split(sharing.SHARING_SEPARATOR_GROUP)[0]
groups_of_user = User.split(sharing.SHARING_SEPARATOR_GROUP)[1].split(',')
Groups = row['User'].removeprefix(sharing.SHARING_SEPARATOR_GROUP).split(',')
logger.trace("sharing/list/check/groups: groups_of_user=%r Groups=%r", groups_of_user, Groups)
found = False
for group in groups_of_user:
if group in Groups:
found = True
break
if found:
pass pass
else: else:
logger.trace("sharing/list/row: add: %r", row) continue
result.append(row) elif row['User'] == User:
pass
else:
continue
if group_check and User.endswith(row['User']):
pass
elif row['User'] == User:
pass
else:
continue
row_copy = row.copy()
if group_check and User is not None:
if row['User'].startswith(sharing.SHARING_SEPARATOR_GROUP):
user_without_group = User.split(sharing.SHARING_SEPARATOR_GROUP)[0]
else:
user_without_group = User
row_copy['PathOrToken'] = row['PathOrToken'].replace("{user}", user_without_group) # replace placeholder
row_copy['User'] = user_without_group # replace with real user
if PathOrToken is not None and row_copy['PathOrToken'] != PathOrToken:
continue
logger.trace("sharing/list/row: add : %r", row_copy)
result.append(row_copy)
return result return result

View File

@@ -149,6 +149,62 @@ class TestBaseAuthRequests(BaseTest):
check = 401 check = 401
self._test_htpasswd("plain", "😀:🔑", "unicode", check=check) self._test_htpasswd("plain", "😀:🔑", "unicode", check=check)
def test_htpasswd_invalid_user_start_with_at(self) -> None:
"""user start with @ is not permitted"""
self._test_htpasswd("plain", "@domain.example:test", (
("@domain.example", "test", True), ("@domain.example", "test", False)), check=401)
def test_htpasswd_invalid_user_end_with_at(self) -> None:
"""user end with @ is not permitted"""
self._test_htpasswd("plain", "domain.example@:test", (
("domain.example@", "test", True), ("domain.example@", "test", False)), check=401)
def test_htpasswd_invalid_user_start_with_encoded_at(self) -> None:
"""user start with encoded @ is not permitted"""
self.configure({"auth": {"urldecode_username": "True"}})
self._test_htpasswd("plain", "@domain.example:test", (
("%40domain.example", "test", True), ("%40domain.example", "test", False)), check=401)
def test_htpasswd_invalid_user_end_with_encoded_at(self) -> None:
"""user end with encoded @ is not permitted"""
self.configure({"auth": {"urldecode_username": "True"}})
self._test_htpasswd("plain", "domain.example@:test", (
("domain.example%40", "test", True), ("domain.example%40", "test", False)), check=401)
def test_htpasswd_invalid_user_with_more_encoded_at(self) -> None:
"""user with more encoded @ is not permitted"""
self.configure({"auth": {"urldecode_username": "True"}})
self._test_htpasswd("plain", "user@group@domain.example:test", (
("user%40group%40domain.example", "test", True), ("user%40group%40domain.example", "test", False)), check=401)
def test_htpasswd_invalid_user_start_with_colon(self) -> None:
"""user start with : is not permitted"""
try:
self._test_htpasswd("plain", ":group:test", (
(":group", "test", True), (":group", "test", False)), check=401)
except RuntimeError:
pass
else:
raise
def test_htpasswd_invalid_user_start_with_encoded_colon(self) -> None:
"""user start with encoded : is not permitted"""
self.configure({"auth": {"urldecode_username": "True"}})
self._test_htpasswd("plain", "'%3Adomain.example:test", (
("%3Adomain.example", "test", True), ("%3Adomain.example", "test", False)), check=401)
def test_htpasswd_invalid_user_end_with_encoded_colon(self) -> None:
"""user end with encoded : is not permitted"""
self.configure({"auth": {"urldecode_username": "True"}})
self._test_htpasswd("plain", "domain.example:test", (
("domain.example%3A", "test", True), ("domain.example%3A", "test", False)), check=401)
def test_htpasswd_invalid_user_with_any_encoded_colon(self) -> None:
"""user with any encoded : is not permitted"""
self.configure({"auth": {"urldecode_username": "True"}})
self._test_htpasswd("plain", "user%3Adomain.example:test", (
("user%3Adomain.example", "test", True), ("user%3Adomain.example", "test", False)), check=401)
def test_htpasswd_md5(self) -> None: def test_htpasswd_md5(self) -> None:
self._test_htpasswd("md5", "tmp:$apr1$BI7VKCZh$GKW4vq2hqDINMr8uv7lDY/") self._test_htpasswd("md5", "tmp:$apr1$BI7VKCZh$GKW4vq2hqDINMr8uv7lDY/")

View File

@@ -0,0 +1,140 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
"""
Radicale tests related to group lookup.
"""
import logging
import os
import sys
import pytest
import radicale
from radicale.tests import BaseTest
class TestBaseGroupRequests(BaseTest):
"""Tests basic requests with group lookup.
We should setup auth for each type before creating the Application object.
"""
def _test_htgroup(self, htpasswd_content: str, htgroup_content, check: int = 207) -> None:
"""Test htpasswd authentication with user "tmp" and password "bepo" for
"""
htpasswd_file_path = os.path.join(self.colpath, ".htpasswd")
htgroup_file_path = os.path.join(self.colpath, ".htgroup")
encoding: str = self.configuration.get("encoding", "stock")
with open(htpasswd_file_path, "w", encoding=encoding) as f:
f.write(htpasswd_content)
with open(htgroup_file_path, "w", encoding=encoding) as f:
f.write(htgroup_content)
self.configure({"auth": {"type": "htpasswd",
"delay": 0,
"htpasswd_filename": htpasswd_file_path,
"htpasswd_encryption": "autodetect"},
"group": {"type": "htgroup",
"htgroup_filename": htgroup_file_path},
"server": {"delay_on_error": 0}})
self.propfind("/", check=check,
login="%s:%s" % ("tmp", "bepo"))
@pytest.mark.skipif(radicale.log.logger.getEffectiveLevel() == logging.INFO, reason="requires loglevel DEBUG")
def test_htgroup_simple(self, caplog) -> None:
caplog.set_level(logging.DEBUG)
self._test_htgroup(htpasswd_content="tmp:bepo",
htgroup_content="group:tmp")
logs = caplog.messages
assert len([log for log in logs if "Group memberships (htgroup) for user 'tmp': {'group'}" in log]) == 1
@pytest.mark.skipif(radicale.log.logger.getEffectiveLevel() == logging.INFO, reason="requires loglevel DEBUG")
def test_htgroup_more_groups(self, caplog) -> None:
caplog.set_level(logging.DEBUG)
self._test_htgroup(htpasswd_content="tmp:bepo",
htgroup_content="group1:tmp\ngroup2:tmp\ngroup3:user")
logs = caplog.messages
assert len([log for log in logs
if "Group memberships (htgroup) for user 'tmp': {'group2', 'group1'}" in log
or "Group memberships (htgroup) for user 'tmp': {'group1', 'group2'}" in log
]) == 1
@pytest.mark.skipif(radicale.log.logger.getEffectiveLevel() == logging.INFO, reason="requires loglevel DEBUG")
def test_htgroup_more_empty_groups(self, caplog) -> None:
caplog.set_level(logging.DEBUG)
self._test_htgroup(htpasswd_content="tmp:bepo",
htgroup_content="group1:tmp\ngroup2:tmp\ngroup3:user\ngroup4:")
logs = caplog.messages
assert len([log for log in logs
if "Group memberships (htgroup) for user 'tmp': {'group2', 'group1'}" in log
or "Group memberships (htgroup) for user 'tmp': {'group1', 'group2'}" in log
]) == 1
@pytest.mark.skipif(radicale.log.logger.getEffectiveLevel() == logging.INFO, reason="requires loglevel DEBUG")
def test_htgroup_more_users(self, caplog) -> None:
caplog.set_level(logging.DEBUG)
self._test_htgroup(htpasswd_content="tmp:bepo",
htgroup_content="group1:tmp user1\ngroup2:tmp user2\ngroup3:user3 user2")
logs = caplog.messages
assert len([log for log in logs
if "Group memberships (htgroup) for user 'tmp': {'group2', 'group1'}" in log
or "Group memberships (htgroup) for user 'tmp': {'group1', 'group2'}" in log
]) == 1
@pytest.mark.skipif(radicale.log.logger.getEffectiveLevel() == logging.INFO, reason="requires loglevel DEBUG")
def test_htgroup_unauthenticated_user(self, caplog) -> None:
caplog.set_level(logging.DEBUG)
self._test_htgroup(htpasswd_content="tmp:bepo1",
htgroup_content="group1:tmp user1\ngroup2:tmp user2\ngroup3:user3 user2", check=401)
logs = caplog.messages
assert len([log for log in logs
if "Group memberships (htgroup) for user 'tmp': {'group2', 'group1'}" in log
or "Group memberships (htgroup) for user 'tmp': {'group1', 'group2'}" in log
]) == 0
@pytest.mark.skipif(sys.platform == "darwin" or sys.platform == 'win32', reason="not supported on MacOS or Windows")
def test_incompatible_group_from_auth(self) -> None:
for auth_type in ["dovecot", "imap", "remote_user", "http_remote_user", "htpasswd", "oauth2"]:
logging.info("\n*** test: auth_type=%r, group_type=%r", "dovecot", auth_type)
try:
self.configure(
{"auth": {
"type": auth_type,
"oauth2_token": "dummy",
},
"group": {"type": "from_auth"}
})
except RuntimeError:
pass
else:
raise
for auth_type in ["pam", "ldap"]:
logging.info("\n*** test: auth_type=%r, group_type=%r", "dovecot", auth_type)
try:
self.configure(
{"auth": {
"type": auth_type,
},
"group": {"type": "from_auth"}
})
except RuntimeError:
raise
else:
pass

View File

@@ -48,6 +48,7 @@ class TestSharingApiSanity(BaseTest):
def setup_method(self) -> None: def setup_method(self) -> None:
BaseTest.setup_method(self) BaseTest.setup_method(self)
self.htpasswd_file_path = os.path.join(self.colpath, ".htpasswd") self.htpasswd_file_path = os.path.join(self.colpath, ".htpasswd")
self.htgroup_file_path = os.path.join(self.colpath, ".htgroup")
encoding: str = self.configuration.get("encoding", "stock") encoding: str = self.configuration.get("encoding", "stock")
htpasswd = ["owner:ownerpw", "user:userpw", htpasswd = ["owner:ownerpw", "user:userpw",
"owner1:owner1pw", "user1:user1pw", "owner1:owner1pw", "user1:user1pw",
@@ -55,11 +56,25 @@ class TestSharingApiSanity(BaseTest):
"owner.surename@domain.example:owner@pw", "user.surename@domain.example:user@pw", "owner.surename@domain.example:owner@pw", "user.surename@domain.example:user@pw",
"owner-surename@domain.example:owner@pw", "user-surename@domain.example:user@pw", "owner-surename@domain.example:owner@pw", "user-surename@domain.example:user@pw",
"owner_surename@domain.example:owner@pw", "user_surename@domain.example:user@pw", "owner_surename@domain.example:owner@pw", "user_surename@domain.example:user@pw",
"user1@domain.example:user1@pw", "user2@domain.example:user2@pw",
"user3:user3pw", "user4:user4",
"user1@domain.tld:user1@pw", "user2@domain.tld:user2@pw",
"us😀er:user😀pw", "us😀er:user😀pw",
"owner2:owner2pw", "user2:user2pw"] "owner2:owner2pw", "user2:user2pw"]
htgroup = ["group1:user1",
"group2:user2",
"group3:user3",
"group4:user4",
"group12:user1 user2",
"group13:user1 user3",
"group23:user2 user3",
]
htpasswd_content = "\n".join(htpasswd) htpasswd_content = "\n".join(htpasswd)
htgroup_content = "\n".join(htgroup)
with open(self.htpasswd_file_path, "w", encoding=encoding) as f: with open(self.htpasswd_file_path, "w", encoding=encoding) as f:
f.write(htpasswd_content) f.write(htpasswd_content)
with open(self.htgroup_file_path, "w", encoding=encoding) as f:
f.write(htgroup_content)
# Helper functions # Helper functions
def _sharing_api(self, sharing_type: str, action: str, check: int, login: Union[str, None], data: str, content_type: str, accept: Union[str, None], x_forwarded_for: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]: def _sharing_api(self, sharing_type: str, action: str, check: int, login: Union[str, None], data: str, content_type: str, accept: Union[str, None], x_forwarded_for: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
@@ -114,7 +129,7 @@ class TestSharingApiSanity(BaseTest):
assert status == 200 assert status == 200
return prop.text return prop.text
def _proppatch_calendar_color(self, path, login, color) -> None: def _proppatch_calendar_color(self, path, login, color, check=207) -> None:
_, responses = self.proppatch(path=path, data="""\ _, responses = self.proppatch(path=path, data="""\
<?xml version="1.0" encoding="utf-8"?> <?xml version="1.0" encoding="utf-8"?>
<D:propertyupdate xmlns:D="DAV:"> <D:propertyupdate xmlns:D="DAV:">
@@ -123,7 +138,9 @@ class TestSharingApiSanity(BaseTest):
<I:calendar-color xmlns:I="http://apple.com/ns/ical/">""" + color + """</I:calendar-color> <I:calendar-color xmlns:I="http://apple.com/ns/ical/">""" + color + """</I:calendar-color>
</D:prop> </D:prop>
</D:set> </D:set>
</D:propertyupdate>""", login=login) </D:propertyupdate>""", login=login, check=check)
if check != 207:
return
logging.info("response: %r", responses) logging.info("response: %r", responses)
response = responses[path] response = responses[path]
assert not isinstance(response, int) and len(response) == 1 assert not isinstance(response, int) and len(response) == 1
@@ -6956,3 +6973,356 @@ permissions: RrWw""")
json_dict['Enabled'] = True json_dict['Enabled'] = True
json_dict['Hidden'] = False json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict) _, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict)
def test_sharing_api_map_user_group_by_domain(self) -> None:
"""share-by-map API usage tests related user group by domain."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_map": "True",
"permit_create_token": "False",
"collection_by_map": "True",
"collection_by_token": "False"},
"logging": {"request_header_on_debug": "False",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"rights": {"type": "owner_only"}})
json_dict: dict
logging.info("\n*** prepare and test access")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
path_mapped = "/owner/calendarPFP-" + db_type + ".ics/"
path_mapped2 = "/owner/calendarPFP2-" + db_type + ".ics/"
path_shared_r = "/{user}/calendarPFP-shared-by-owner-r-" + db_type + ".ics/"
path_shared2_r = "/{user}/calendarPFP2-shared-by-owner-r-" + db_type + ".ics/"
path_shared_r_base = "/{user}/"
self.mkcalendar(path_mapped, login="owner:ownerpw")
self.mkcalendar(path_mapped2, login="owner:ownerpw")
# create map
logging.info("\n*** create map @domain/owner:rP -> success")
json_dict = {}
json_dict['User'] = "@domain.example"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
# create map
logging.info("\n*** create map @domain/owner:rP -> success")
json_dict = {}
json_dict['User'] = "@domain.example"
json_dict['PathMapped'] = path_mapped2
json_dict['PathOrToken'] = path_shared2_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
# verify PROPFIND as user1
logging.info("\n*** PROPFIND collection user1@domain.example")
path_shared_r_user = path_shared_r.replace("{user}", "user1@domain.example")
_, responses = self.propfind(path_shared_r_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user1@domain.example:user1@pw")
assert path_shared_r_user.replace('@', '%40') in responses
# verify PROPFIND as user2
logging.info("\n*** PROPFIND collection user2@domain.example")
path_shared_r_user = path_shared_r.replace("{user}", "user2@domain.example")
_, responses = self.propfind(path_shared_r_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user2@domain.example:user2@pw")
assert path_shared_r_user.replace('@', '%40') in responses
# verify PROPFIND as user1
logging.info("\n*** PROPFIND collection user1@domain.tld")
path_shared_r_user = path_shared_r.replace("{user}", "user1@domain.tld")
_, responses = self.propfind(path_shared_r_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user1@domain.tld:user1@pw", check=404)
# verify PROPFIND as user2
logging.info("\n*** PROPFIND collection user2@domain.tld")
path_shared_r_user = path_shared_r.replace("{user}", "user2@domain.tld")
_, responses = self.propfind(path_shared_r_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user2@domain.tld:user2@pw", check=404)
# verify PROPFIND as user1 in list
logging.info("\n*** PROPFIND collection DEPTH=1 user1@domain.example")
path_shared_r_base_user = path_shared_r_base.replace("{user}", "user1@domain.example")
path_shared_r_user = path_shared_r.replace("{user}", "user1@domain.example")
path_shared2_r_user = path_shared_r.replace("{user}", "user1@domain.example")
_, responses = self.propfind(path_shared_r_base_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user1@domain.example:user1@pw", HTTP_DEPTH="1")
assert path_shared_r_base_user.replace('@', '%40') in responses
assert path_shared_r_user.replace('@', '%40') in responses
assert path_shared2_r_user.replace('@', '%40') in responses
# execute PROPPATCH as user
logging.info("\n*** PROPPATCH collection user1@domain.example -> forbidden")
self._proppatch_calendar_color(path_shared_r_user, login="user1@domain.example:user1@pw", color="#FFFFFF", check=403)
# verify PROPFIND as user1 not in list
logging.info("\n*** PROPFIND collection DEPTH=1 user1@domain.tld")
path_shared_r_base_user = path_shared_r_base.replace("{user}", "user1@domain.tld")
path_shared_r_user = path_shared_r.replace("{user}", "user1@domain.tld")
_, responses = self.propfind(path_shared_r_base_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user1@domain.tld:user1@pw", HTTP_DEPTH="1")
assert path_shared_r_base_user.replace('@', '%40') in responses
assert path_shared_r_user.replace('@', '%40') not in responses
logging.info("\n*** PROPPATCH collection user1@domain.tld -> not found")
self._proppatch_calendar_color(path_shared_r_user, login="user1@domain.tld:user1@pw", color="#FFFFFF", check=404)
def test_sharing_api_map_user_group_by_local(self) -> None:
"""share-by-map API usage tests related user group by local."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"group": {"type": "htgroup",
"htgroup_filename": self.htgroup_file_path},
"sharing": {
"type": "csv",
"permit_create_map": "True",
"permit_create_token": "False",
"collection_by_map": "True",
"collection_by_token": "False"},
"logging": {"request_header_on_debug": "False",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"rights": {"type": "owner_only"}})
json_dict: dict
logging.info("\n*** prepare and test access")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
path_mapped1 = "/owner/calendarUGBL1-" + db_type + ".ics/"
path_mapped2 = "/owner/calendarUGBL2-" + db_type + ".ics/"
path_mapped3 = "/owner/calendarUGBL3-" + db_type + ".ics/"
path_shared1_r = "/{user}/calendarUGBL1-shared-by-owner-r-" + db_type + ".ics/"
path_shared2_r = "/{user}/calendarUGBL2-shared-by-owner-r-" + db_type + ".ics/"
path_shared3_r = "/{user}/calendarUGBL3-shared-by-owner-r-" + db_type + ".ics/"
path_shared_r_base = "/{user}/"
self.mkcalendar(path_mapped1, login="owner:ownerpw")
self.mkcalendar(path_mapped2, login="owner:ownerpw")
self.mkcalendar(path_mapped3, login="owner:ownerpw")
# create map
logging.info("\n*** create map :group1/owner -> 400 (unsupported permissions)")
json_dict = {}
json_dict['User'] = ":group1"
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "rP"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** create map :group1/owner -> 400 (unsupported permissions)")
json_dict = {}
json_dict['User'] = ":group1"
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "rE"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** create map :group1/owner -> 400 (unsupported permissions)")
json_dict = {}
json_dict['User'] = ":group1"
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "rEP"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** create map :group1/owner -> success")
json_dict = {}
json_dict['User'] = ":group1"
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** create map :group2/owner -> success")
json_dict = {}
json_dict['User'] = ":group2"
json_dict['PathMapped'] = path_mapped2
json_dict['PathOrToken'] = path_shared2_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** create map :group1,group2/owner -> success")
json_dict = {}
json_dict['User'] = ":group1,group2"
json_dict['PathMapped'] = path_mapped3
json_dict['PathOrToken'] = path_shared3_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
# verify PROPFIND as user1 in list
logging.info("\n*** PROPFIND collection DEPTH=1 user1")
path_shared_r_base_user = path_shared_r_base.replace("{user}", "user1")
path_shared1_r_user = path_shared1_r.replace("{user}", "user1")
path_shared2_r_user = path_shared2_r.replace("{user}", "user1")
path_shared3_r_user = path_shared3_r.replace("{user}", "user1")
_, responses = self.propfind(path_shared_r_base_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user1:user1pw", HTTP_DEPTH="1")
assert path_shared_r_base_user in responses
assert path_shared1_r_user in responses
assert path_shared2_r_user not in responses
assert path_shared3_r_user in responses
# verify PROPFIND as user2 in list
logging.info("\n*** PROPFIND collection DEPTH=1 user2")
path_shared_r_base_user = path_shared_r_base.replace("{user}", "user2")
path_shared1_r_user = path_shared1_r.replace("{user}", "user2")
path_shared2_r_user = path_shared2_r.replace("{user}", "user2")
path_shared3_r_user = path_shared3_r.replace("{user}", "user2")
_, responses = self.propfind(path_shared_r_base_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user2:user2pw", HTTP_DEPTH="1")
assert path_shared_r_base_user in responses
assert path_shared1_r_user not in responses
assert path_shared2_r_user in responses
assert path_shared3_r_user in responses
# try upload item as user1 -> fail (w permission missing)
logging.info("\n*** PUT to shared1 as user1 -> 403")
path_shared1_r_user = path_shared1_r.replace("{user}", "user1")
event = get_file_content("event1.ics")
self.put(path_shared1_r_user, event, login="user1:user1pw", check=403)
# update permissions
logging.info("\n*** update map :group1/owner -> success")
json_dict = {}
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "rw"
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map :group3/owner -> success")
json_dict = {}
json_dict['PathMapped'] = path_mapped3
json_dict['PathOrToken'] = path_shared3_r
json_dict['Permissions'] = "rw"
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
# upload item as user1 -> success
logging.info("\n*** PUT to shared1 as user1 -> 201")
path_shared1_r_user = path_shared1_r.replace("{user}", "user1")
event = get_file_content("event1.ics")
self.put(path_shared1_r_user, event, login="user1:user1pw")
# propfind as user1 -> success
logging.info("\n*** PROPFIND collection user1")
path_shared1_r_user = path_shared1_r.replace("{user}", "user1")
_, responses = self.propfind(path_shared1_r_user, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<calendar-home-set xmlns="urn:ietf:params:xml:ns:caldav" />
</propfind>""", login="user1:user1pw")
assert path_shared1_r_user in responses
# report as user1 -> success
logging.info("\n*** REPORT collection user1")
path_shared1_r_user = path_shared1_r.replace("{user}", "user1")
item_shared1_r_user = path_shared1_r.replace("{user}", "user1") + "event1.ics"
_, responses = self.report(path_shared1_r_user, """\
<?xml version="1.0" encoding="utf-8" ?>
<C:calendar-query xmlns:C="urn:ietf:params:xml:ns:caldav">
<D:prop xmlns:D="DAV:">
<D:getetag />
</D:prop>
</C:calendar-query>""", login="user1:user1pw")
assert item_shared1_r_user in responses
# report as user2 -> success
logging.info("\n*** REPORT collection user2")
path_shared1_r_user = path_shared1_r.replace("{user}", "user2")
item_shared1_r_user = path_shared1_r.replace("{user}", "user2") + "event1.ics"
_, responses = self.report(path_shared1_r_user, """\
<?xml version="1.0" encoding="utf-8" ?>
<C:calendar-query xmlns:C="urn:ietf:params:xml:ns:caldav">
<D:prop xmlns:D="DAV:">
<D:getetag />
</D:prop>
</C:calendar-query>""", login="user2:user2pw", check=404)
# get item as user1 -> success
logging.info("\n*** GET from shared1 as user1")
item_shared1_r_user = path_shared1_r.replace("{user}", "user1") + "event1.ics"
self.get(item_shared1_r_user, login="user1:user1pw")
# get item as user2 -> 404
logging.info("\n*** GET from shared3 as user2")
item_shared3_r_user = path_shared1_r.replace("{user}", "user2") + "event1.ics"
self.get(item_shared3_r_user, login="user2:user2pw", check=404)
# move item as user1 -> success
logging.info("\n*** MOVE item shared1 to shared3 as user1")
item_shared1_r_user = path_shared1_r.replace("{user}", "user1") + "event1.ics"
item_shared3_r_user = path_shared3_r.replace("{user}", "user1") + "event1.ics"
self.request("MOVE", item_shared1_r_user, login="user1:user1pw", HTTP_DESTINATION="http://127.0.0.1"+item_shared3_r_user)
# get item as user2 -> 200
logging.info("\n*** GET from shared3 as user2")
item_shared3_r_user = path_shared3_r.replace("{user}", "user2") + "event1.ics"
self.get(item_shared3_r_user, login="user2:user2pw")
# delete item as user1 -> 404
logging.info("\n*** DELETE from shared1 as user1 -> 404")
item_shared1_r_user = path_shared1_r.replace("{user}", "user1") + "event1.ics"
self.delete(item_shared1_r_user, login="user1:user1pw", check=404)
# delete item as user1
logging.info("\n*** DELETE from shared3 as user1 -> 200")
item_shared3_r_user = path_shared3_r.replace("{user}", "user1") + "event1.ics"
self.delete(item_shared3_r_user, login="user1:user1pw")
# try proppatch -> 403
logging.info("\n*** PROPPATCH shared3 as user1 -> 403")
path_shared3_r_user = path_shared3_r.replace("{user}", "user1")
self._proppatch_calendar_color(path_shared3_r_user, login="user1:user1pw", color="#FFFFFF", check=403)

View File

@@ -45,6 +45,7 @@ install_requires = ["defusedxml", "libpass>=1.9.3", "vobject>=0.9.6",
bcrypt_requires = ["bcrypt"] bcrypt_requires = ["bcrypt"]
argon2_requires = ["argon2-cffi"] argon2_requires = ["argon2-cffi"]
ldap_requires = ["ldap3"] ldap_requires = ["ldap3"]
pam_requires = ["pam"]
test_requires = ["pytest>=7", "waitress", *bcrypt_requires, *argon2_requires] test_requires = ["pytest>=7", "waitress", *bcrypt_requires, *argon2_requires]
setup( setup(
@@ -63,7 +64,7 @@ setup(
package_data={"radicale": [*web_files, "py.typed"]}, package_data={"radicale": [*web_files, "py.typed"]},
entry_points={"console_scripts": ["radicale = radicale.__main__:run"]}, entry_points={"console_scripts": ["radicale = radicale.__main__:run"]},
install_requires=install_requires, install_requires=install_requires,
extras_require={"test": test_requires, "bcrypt": bcrypt_requires, "argon2": argon2_requires, "ldap": ldap_requires}, extras_require={"test": test_requires, "bcrypt": bcrypt_requires, "argon2": argon2_requires, "ldap": ldap_requires, "pam": pam_requires},
keywords=["calendar", "addressbook", "CalDAV", "CardDAV"], keywords=["calendar", "addressbook", "CalDAV", "CardDAV"],
python_requires=">=3.9.0", python_requires=">=3.9.0",
classifiers=[ classifiers=[