Merge branch 'master' into deprecate-ldap_use_ssl

This commit is contained in:
Peter Bieringer
2026-02-24 21:33:30 +01:00
committed by GitHub
28 changed files with 4723 additions and 72 deletions

View File

@@ -1,6 +1,12 @@
# Changelog
## 3.6.1.dev
## 3.7.0.dev
* Feature: collection sharing by map or token incl. management API
* Add: [auth] oauth2: allow custom client_id and client_secret (optional)
* Cleanup: deprecate config option 'ldap_use_ssl' for good
## 3.6.1
* Fix: MOVE failing with URL-encoded destination header
* Improve: add workaround to remove empty lines in item to avoid reject by vobject parser
@@ -8,7 +14,7 @@
* Fix: RECURRENCE-ID comparison on all-day events
* Fix: format_ut problem on 32-bit systems (got lost inbetween since fixed in 3.5.8)
* Maintain: Refactor Web UI Javascript into separate files
* Cleanup: deprecate config option 'ldap_use_ssl' for good
* Maintain: code preparation for upcoming "sharing" extension and optimizations
## 3.6.0

View File

@@ -726,6 +726,12 @@ _(>= 3.6.0)_
Verification of a particular item file
##### --verify-sharing
_(>= 3.7.0)_
Verification of local sharing database
##### -C|--config <file>
Load one or more specified config file(s)
@@ -1348,6 +1354,22 @@ Endpoint URL for the OAuth2 token
Default: (unset)
##### oauth2_client_id
_(>= 3.7.0)_
Client ID used to request the Auth2 token
Default: `radicale`
##### oauth2_client_secret
_(>= 3.7.0)_
Client secret used to request the Auth2 token
Default: (unset)
##### pam_service
_(>= 3.5.0)_
@@ -2037,6 +2059,49 @@ is thrown instead of returning the results.
Default: 10000
#### [sharing]
_(>= 3.7.0)_
##### type
_(>= 3.7.0)_
Sharing database type
One of:
* `none`
* `csv`
* `files`
Default: `none` (implicit disabling the feature)
##### database_path
_(>= 3.7.0)_
Sharing database path
Default:
* type `csv`: `(filesystem_folder)/collection-db/sharing.csv`
* type `files`: `(filesystem_folder)/collection-db/files`
##### collection_by_token
_(>= 3.7.0)_
Share collection by token
Default: `false`
##### collection_by_map
_(>= 3.7.0)_
Share collection by map
Default: `false`
## Supported Clients
Radicale has been tested with:

285
SHARING.md Normal file
View File

@@ -0,0 +1,285 @@
# Collection Sharing
Static collection sharing without permissions filter using soft-links (Unix-only) is supported since storage type `multifilesystem` was implemented, see (Wiki: Sharing Collections)[https://github.com/Kozea/Radicale/wiki/Sharing-Collections]
With 3.7.0 a major extension was implemented using internal mapping configuration stored in a database and a management API.
## Sharing Configuration Store
Types of supported sharing configuration:
* csv (_>= 3.7.0_)
* files (_>= 3.7.0_)
### Sharing Configuration Entry Data
* `ShareType`: type of share
* `token`: token-based share (do not require user authentication)
* `map`: map-based share (requires user authentication)
* `PathOrToken`: token or "virtual" collection, has to be unique (PRIMARY KEY)
* `PathMapped`: target collection
* `Owner`: owner of the share
* `User`: user of the share
* `Permissions`: effective permission of the share
* `EnabledByOwner`: control by owner
* `EnabledByUser`: control by user
* `HiddenByOwner`: control by owner
* `HiddenByUser`: control by user
* `TimestampCreated`: unixtime of creation
* `TimestampUpdated`: unixtime of last update
`Enabled*`: owner AND user have to enable a share to become usable
`Hidden*`: owner AND user have to disable a share to become visible in PROPFIND
### Sharing Configuration Entry Storage
#### CSV
(_>= 3.7.0_)
One CSV file containing one row per sharing config, separated by `,` and containing header with columns from above.
#### Files
(_>= 3.7.0_)
File-based configuration store is using encoded `PathOrToken` as filename for each config. File contains the data stored as "dict" in binary Python "pickle" format (same is also used for item cache files).
## Sharing Access
### Sharing Access via Maps
(_>= 3.7.0_)
Map-based sharing can be accessed as usual after authentication and authorization.
#### Workflow
* create map as owner
* enable map as owner (can be combined with "create")
* enable map as user (explicit required to avoid sudden available share)
In case share should be visible using PROPFIND
* unhide map as owner (can be combined with "create")
* unhide map as user (explicit required to avoid sudden visible share)
### Sharing Access via Tokens
(_>= 3.7.0_)
Token-based sharing can be accessed after retrieving the token via
Token-URI: `/.token/<Token>`
#### Workflow
* create token as owner
* enable token as owner (can be combined with "create")
* handover URI with token to client
## Sharing Configuration Management API
### Sharing Configuration Management API version 1
(_>= 3.7.0_)
Type: POST API
Base-URI: `/.sharing/v1/<ShareType>/<Hook>`
See also test cases in `radicale/tests/test_sharing.py`
#### Data Format
##### Input Data Format
Parsing be controlled by `CONTENT_TYPE`
* application/x-www-form-urlencoded (_>= 3.7.0_)
* application/json (_>= 3.7.0_)
##### Output Data Format
Can be selected by `HTTP_ACCEPT`
* text/plain (_>= 3.7.0_)
* text/csv (_>= 3.7.0_) - only for "list"
* application/json (_>= 3.7.0_)
##### Accepted Input Data Fields
* `PathOrToken`: token or "virtual" collection
* `PathMapped`: target collection
* `Owner`: owner of the share
* `User`: user of the share
* `Permissions`: effective permission of the share
* `Enabled`: owner/user selected by authentication
* `Hidden`: owner/user selected by authentication
#### API Hooks
##### API Hook "info"
Shows what kind of ShareTypes are supported
* Example: TEXT
```
curl -u user:pass -H "accept: text/plain" -d "" http://localhost:5232/.sharing/v1/all/info
ApiVersion=1
Status=success
FeatureEnabledCollectionByMap=True
PermittedCreateCollectionByMap=True
FeatureEnabledCollectionByToken=True
PermittedCreateCollectionByToken=True
```
* Example: JSON
```
curl -u user:pass --silent -H "accept: application/json" -d "" http://localhost:5232/.sharing/v1/all/info | jq
{
"ApiVersion": 1,
"Status": "success",
"FeatureEnabledCollectionByMap": true,
"PermittedCreateCollectionByMap": true,
"FeatureEnabledCollectionByToken": true,
"PermittedCreateCollectionByToken": true
}
```
##### API Hook "*/create"
Create a share
###### API Hook "token/create"
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathMapped | yes | |
| Permissions | no | r
* Output
| Parameter | Value |
| - | - |
| PathOrToken | (autogenerated token) |
* Example: TEXT
```
curl -u user:pass -d "PathMapped=/user/testcalendar1/" http://localhost:5232/.sharing/v1/token/create
ApiVersion=1
Status=success
PathOrToken=v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw=
```
###### API Hook "map/create"
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | |
| PathMapped | yes | |
| Permissions | no | r
| User | yes | |
* Output: result status
* Example: TEXT
```
curl -u owner:pass -d "PathOrToken=/user/cal1-from-owner/" -d "PathMapped=/owner/cal1/" -d "User=user" http://localhost:5232/.sharing/v1/map/create
ApiVersion=1
Status=success
```
##### API Hook "*/list"
List shares (optional with filter)
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathMapped | no | (all) |
| Owner | no | (owned ones) |
| User | no | (filtered) |
* Output: plain/csv/json
* Example: CSV
```
curl -H "accept: text/csv" -u owner:pass -d "" http://localhost:5232/.sharing/v1/map/list
ShareType,PathOrToken,PathMapped,Owner,User,Permissions,EnabledByOwner,EnabledByUser,HiddenByOwner,HiddenByUser,TimestampCreated,TimestampUpdated
map,/user/cal1-from-owner/,/owner/cal1/,owner,user,r,False,False,True,True,1771962120,1771962120
```
##### API Hook "*/delete"
Delete a share selected by `PathOrToken`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | |
* Output: result status
##### API Hook "*/update"
Update a share selected by `PathOrToken`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | n/a |
| PathMapped | no | |
| User | no | |
* Output: result status
##### API Hooks "*/(enable|disable|hide|unhide)"
Enable|disable|hide|unhide a share selected by `PathOrToken`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | n/a |
* Output: result status
* Example: TEXT (enable)
```
curl -u owner:pass -d "PathOrToken=/user/cal1-from-owner/" -d "PathMapped=/owner/cal1/" -d "User=user" http://localhost:5232/.sharing/v1/map/enable
ApiVersion=1
Status=success
```
* Example: JSON (unhide)
```
curl -u owner:pass -d '{"PathOrToken": "/user/cal1-from-owner/", "PathMapped": "/owner/cal1/", "User": "user"} http://localhost:5232/.sharing/v1/map/unhide
ApiVersion=1
Status=success
```

25
config
View File

@@ -153,6 +153,12 @@
# OAuth2 token endpoint URL
#oauth2_token_endpoint = <URL>
# OAuth2 client id
#oauth2_client_id = radicale
# OAuth2 client secret (optional)
#oauth2_client_secret =
# PAM service
#pam_serivce = radicale
@@ -293,6 +299,25 @@
#predefined_collections =
[sharing]
# Sharing database type
# Value: none | csv | files
#type = none
# Sharing database path for type 'csv'
#database_path = (filesystem_folder)/collection-db/sharing.csv
# Sharing database path for type 'files'
#database_path = (filesystem_folder)/collection-db/files
# Share collection by map
#collection_by_map = false
# Share collection by token
#collection_by_token = false
[web]
# Web interface backend

View File

@@ -3,7 +3,7 @@ name = "Radicale"
# When the version is updated, a new section in the CHANGELOG.md file must be
# added too.
readme = "README.md"
version = "3.6.1.dev"
version = "3.7.0.dev"
authors = [{name = "Guillaume Ayoub", email = "guillaume.ayoub@kozea.fr"}, {name = "Unrud", email = "unrud@outlook.com"}, {name = "Peter Bieringer", email = "pb@bieringer.de"}]
license = {text = "GNU GPL v3"}
description = "CalDAV and CardDAV Server"
@@ -103,7 +103,7 @@ radicale = [
"web/internal_data/css/icons/edit.svg",
"web/internal_data/css/icons/new.svg",
"web/internal_data/css/icons/upload.svg",
"web/internal_data/fn.js",
"web/internal_data/*.js",
"web/internal_data/index.html",
"py.typed",
]

View File

@@ -1,7 +1,7 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2011-2017 Guillaume Ayoub
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -33,7 +33,8 @@ import sys
from types import FrameType
from typing import List, Optional, cast
from radicale import VERSION, config, item, log, server, storage, types
from radicale import (VERSION, config, item, log, server, sharing, storage,
types)
from radicale.log import logger
@@ -67,6 +68,8 @@ def run() -> None:
help="check the storage for errors and exit")
parser.add_argument("--verify-item", action="store", nargs=1,
help="check the provided item file for errors and exit")
parser.add_argument("--verify-sharing", action="store_true",
help="check the sharing database for errors and exit")
parser.add_argument("-C", "--config",
help="use specific configuration files", nargs="*")
parser.add_argument("-D", "--debug", action="store_const", const="debug",
@@ -209,6 +212,19 @@ def run() -> None:
sys.exit(1)
return
if args_ns.verify_sharing:
logger.info("Verifying sharing database")
try:
sharing_ = sharing.load(configuration)
if not sharing_.verify():
logger.critical("Sharing database verification failed")
sys.exit(1)
except Exception as e:
logger.critical("An exception occurred during sharing database "
"verification: %s", e, exc_info=True)
sys.exit(1)
return
# Create a socket pair to notify the server of program shutdown
shutdown_socket, shutdown_socket_out = socket.socketpair()

View File

@@ -1,6 +1,6 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2020 Unrud <unrud@outlook.com>
# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -17,14 +17,14 @@
import io
import logging
import posixpath
import sys
import xml.etree.ElementTree as ET
from typing import Optional
from typing import Optional, Union
from radicale import (auth, config, hook, httputils, pathutils, rights,
storage, types, utils, web, xmlutils)
sharing, storage, types, utils, web, xmlutils)
from radicale.log import logger
from radicale.rights import intersect
# HACK: https://github.com/tiran/defusedxml/issues/54
import defusedxml.ElementTree as DefusedET # isort:skip
@@ -38,6 +38,7 @@ class ApplicationBase:
_storage: storage.BaseStorage
_rights: rights.BaseRights
_web: web.BaseWeb
_sharing: sharing.BaseSharing
_encoding: str
_max_resource_size: int
_permit_delete_collection: bool
@@ -51,6 +52,7 @@ class ApplicationBase:
self._storage = storage.load(configuration)
self._rights = rights.load(configuration)
self._web = web.load(configuration)
self._sharing = sharing.load(configuration)
self._encoding = configuration.get("encoding", "request")
self._log_bad_put_request_content = configuration.get("logging", "bad_put_request_content")
self._response_content_on_debug = configuration.get("logging", "response_content_on_debug")
@@ -106,15 +108,19 @@ class Access:
permissions: str
_rights: rights.BaseRights
_parent_permissions: Optional[str]
_permissions_filter: Union[str, None] = None
def __init__(self, rights: rights.BaseRights, user: str, path: str
def __init__(self, rights: rights.BaseRights, user: str, path: str, permissions_filter: Union[str, None] = None
) -> None:
self._rights = rights
self.user = user
self.path = path
self.parent_path = pathutils.unstrip_path(
posixpath.dirname(pathutils.strip_path(path)), True)
self.parent_path = pathutils.parent_path(path)
self.permissions = self._rights.authorization(self.user, self.path)
if permissions_filter is not None:
self._permissions_filter = permissions_filter
permissions_filtered = intersect(self.permissions, permissions_filter)
self.permissions = permissions_filtered
self._parent_permissions = None
@property
@@ -124,6 +130,9 @@ class Access:
if self._parent_permissions is None:
self._parent_permissions = self._rights.authorization(
self.user, self.parent_path)
if self._permissions_filter is not None:
parent_permissions_filtered = intersect(self._parent_permissions, self._permissions_filter)
self._parent_permissions = parent_permissions_filtered
return self._parent_permissions
def check(self, permission: str,

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2020 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -57,7 +57,16 @@ class ApplicationPartDelete(ApplicationBase):
def do_DELETE(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage DELETE request."""
access = Access(self._rights, user, path)
permissions_filter = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"):
return httputils.NOT_ALLOWED
with self._storage.acquire_lock("w", user, path=path, request="DELETE"):

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2023 Unrud <unrud@outlook.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-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
@@ -76,7 +76,16 @@ class ApplicationPartGet(ApplicationBase):
return httputils.redirect(location, client.MOVED_PERMANENTLY)
# Dispatch /.web path to web module
return self._web.get(environ, base_prefix, path, user)
access = Access(self._rights, user, path)
permissions_filter = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
if not access.check("r") and "i" not in access.permissions:
return httputils.NOT_ALLOWED
with self._storage.acquire_lock("r", user):

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2021 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -19,7 +19,6 @@
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
import errno
import posixpath
import re
import socket
from http import client
@@ -55,6 +54,13 @@ class ApplicationPartMkcalendar(ApplicationBase):
logger.warning(
"Bad MKCALENDAR request on %r: %s", path, e, exc_info=True)
return httputils.BAD_REQUEST
if self._sharing._enabled:
# check for shared collections (active or inactive)
collections_shared_map = self._sharing.sharing_collection_map_list(user, active=False)
if collections_shared_map:
for sharing in collections_shared_map:
if sharing['PathOrToken'] == path:
return httputils.CONFLICT
# TODO: use this?
# timezone = props.get("C:calendar-timezone")
with self._storage.acquire_lock("w", user, path=path, request="MKCALENDAR"):
@@ -62,8 +68,7 @@ class ApplicationPartMkcalendar(ApplicationBase):
if item:
return self._webdav_error_response(
client.CONFLICT, "D:resource-must-be-null")
parent_path = pathutils.unstrip_path(
posixpath.dirname(pathutils.strip_path(path)), True)
parent_path = pathutils.parent_path(path)
parent_item = next(iter(self._storage.discover(parent_path)), None)
if not parent_item:
return httputils.CONFLICT

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2021 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -19,7 +19,6 @@
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
import errno
import posixpath
import re
import socket
from http import client
@@ -62,12 +61,18 @@ class ApplicationPartMkcol(ApplicationBase):
if not props.get("tag") and "W" not in permissions:
logger.warning("MKCOL request %r (type:%s): %s", path, collection_type, "rejected because of missing rights 'W'")
return httputils.NOT_ALLOWED
if self._sharing._enabled:
# check for shared collections (active or inactive)
collections_shared_map = self._sharing.sharing_collection_map_list(user, active=False)
if collections_shared_map:
for sharing in collections_shared_map:
if sharing['PathOrToken'] == path:
return httputils.CONFLICT
with self._storage.acquire_lock("w", user, path=path, request="MKCOL"):
item = next(iter(self._storage.discover(path)), None)
if item:
return httputils.METHOD_NOT_ALLOWED
parent_path = pathutils.unstrip_path(
posixpath.dirname(pathutils.strip_path(path)), True)
parent_path = pathutils.parent_path(path)
parent_item = next(iter(self._storage.discover(parent_path)), None)
if not parent_item:
return httputils.CONFLICT

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2023 Unrud <unrud@outlook.com>
# Copyright © 2023-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2023-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
@@ -67,7 +67,18 @@ class ApplicationPartMove(ApplicationBase):
# Remote destination server, not supported
return httputils.REMOTE_DESTINATION
access = Access(self._rights, user, path)
to_user = user
to_permissions_filter = None
permissions_filter = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"):
return httputils.NOT_ALLOWED
to_path = pathutils.sanitize_path(to_url.path)
@@ -76,7 +87,16 @@ class ApplicationPartMove(ApplicationBase):
"start with base prefix", to_path, path)
return httputils.NOT_ALLOWED
to_path = to_path[len(base_prefix):]
to_access = Access(self._rights, user, to_path)
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(to_path, to_user)
if sharing:
# overwrite and run through extended permission check
to_path = sharing['PathMapped']
to_user = sharing['Owner']
to_permissions_filter = sharing['Permissions']
to_access = Access(self._rights, to_user, to_path, to_permissions_filter)
to_access = Access(self._rights, to_user, to_path, to_permissions_filter)
if not to_access.check("w"):
return httputils.NOT_ALLOWED
@@ -94,8 +114,7 @@ class ApplicationPartMove(ApplicationBase):
to_item = next(iter(self._storage.discover(to_path)), None)
if isinstance(to_item, storage.BaseCollection):
return httputils.FORBIDDEN
to_parent_path = pathutils.unstrip_path(
posixpath.dirname(pathutils.strip_path(to_path)), True)
to_parent_path = pathutils.parent_path(to_path)
to_collection = next(iter(
self._storage.discover(to_parent_path)), None)
if not to_collection:

View File

@@ -4,7 +4,7 @@
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2021 Unrud <unrud@outlook.com>
# Copyright © 2020-2020 Tom Hacohen <tom@stosb.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-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
@@ -30,4 +30,6 @@ class ApplicationPartPost(ApplicationBase):
"""Manage POST request."""
if path == "/.web" or path.startswith("/.web/"):
return self._web.post(environ, base_prefix, path, user)
elif path == "/.sharing" or path.startswith("/.sharing/"):
return self._sharing.post(environ, base_prefix, path, user)
return httputils.METHOD_NOT_ALLOWED

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2021 Unrud <unrud@outlook.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-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
@@ -20,11 +20,13 @@
import collections
import itertools
import logging
import posixpath
import socket
import xml.etree.ElementTree as ET
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)
from radicale import (httputils, pathutils, rights, storage, types, utils,
xmlutils)
@@ -35,7 +37,7 @@ from radicale.log import logger
def xml_propfind(base_prefix: str, path: str,
xml_request: Optional[ET.Element],
allowed_items: Iterable[Tuple[types.CollectionOrItem, str]],
user: str, encoding: str, max_resource_size: int) -> Optional[ET.Element]:
user: str, encoding: str, max_resource_size: int, sharing: Union[dict, None] = None) -> Optional[ET.Element]:
"""Read and answer PROPFIND requests.
Read rfc4918-9.1 for info.
@@ -72,7 +74,7 @@ def xml_propfind(base_prefix: str, path: str,
write = permission == "w"
multistatus.append(xml_propfind_response(
base_prefix, path, item, props, user, encoding, write=write,
allprop=allprop, propname=propname, max_resource_size=max_resource_size))
allprop=allprop, propname=propname, max_resource_size=max_resource_size, sharing=sharing))
return multistatus
@@ -80,7 +82,7 @@ def xml_propfind(base_prefix: str, path: str,
def xml_propfind_response(
base_prefix: str, path: str, item: types.CollectionOrItem,
props: Sequence[str], user: str, encoding: str, max_resource_size: int, write: bool = False,
propname: bool = False, allprop: bool = False) -> ET.Element:
propname: bool = False, allprop: bool = False, sharing: Union[dict, None] = None) -> ET.Element:
"""Build and return a PROPFIND response."""
if propname and allprop or (props and (propname or allprop)):
raise ValueError("Only use one of props, propname and allprops")
@@ -100,6 +102,9 @@ def xml_propfind_response(
collection.path, item.href))
response = ET.Element(xmlutils.make_clark("D:response"))
href = ET.Element(xmlutils.make_clark("D:href"))
if sharing:
# backmap
uri = uri.replace(sharing['PathMapped'], sharing['PathOrToken'])
href.text = xmlutils.make_href(base_prefix, uri)
response.append(href)
@@ -178,6 +183,9 @@ def xml_propfind_response(
is_collection and collection.is_principal):
child_element = ET.Element(xmlutils.make_clark("D:href"))
child_element.text = xmlutils.make_href(base_prefix, path)
if sharing:
# backmap
child_element.text = child_element.text.replace(sharing['PathMapped'], sharing['PathOrToken'])
element.append(child_element)
elif tag == xmlutils.make_clark("C:supported-calendar-component-set"):
human_tag = xmlutils.make_human_tag(tag)
@@ -213,6 +221,9 @@ def xml_propfind_response(
child_element = ET.Element(xmlutils.make_clark("D:href"))
child_element.text = xmlutils.make_href(
base_prefix, "/%s/" % user)
if sharing:
# backmap
child_element.text = child_element.text.replace(sharing['Owner'], sharing['User'])
element.append(child_element)
else:
element.append(ET.Element(
@@ -373,6 +384,8 @@ class ApplicationPartPropfind(ApplicationBase):
for item in items:
if isinstance(item, storage.BaseCollection):
path = pathutils.unstrip_path(item.path, True)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND/_collect_allowed_items/BaseCollection: path=%r user=%r", path, user)
if item.tag:
permissions = rights.intersect(
self._rights.authorization(user, path), "rw")
@@ -405,7 +418,18 @@ class ApplicationPartPropfind(ApplicationBase):
def do_PROPFIND(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage PROPFIND request."""
access = Access(self._rights, user, path)
http_depth = environ.get("HTTP_DEPTH", "0")
permissions_filter = None
sharing = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
if not access.check("r"):
return httputils.NOT_ALLOWED
try:
@@ -418,8 +442,10 @@ class ApplicationPartPropfind(ApplicationBase):
logger.debug("Client timed out", exc_info=True)
return httputils.REQUEST_TIMEOUT
with self._storage.acquire_lock("r", user):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: discover path=%r depth=%s", path, http_depth)
items_iter = iter(self._storage.discover(
path, environ.get("HTTP_DEPTH", "0"),
path, http_depth,
None, self._rights._user_groups))
# take root item for rights checking
item = next(items_iter, None)
@@ -429,11 +455,36 @@ class ApplicationPartPropfind(ApplicationBase):
return httputils.NOT_ALLOWED
# put item back
items_iter = itertools.chain([item], items_iter)
allowed_items = self._collect_allowed_items(items_iter, user)
headers = {"DAV": httputils.DAV_HEADERS,
"Content-Type": "text/xml; charset=%s" % self._encoding}
xml_answer = xml_propfind(base_prefix, path, xml_content,
allowed_items, user, self._encoding, max_resource_size=self._max_resource_size)
if xml_answer is None:
return httputils.NOT_ALLOWED
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)
allowed_items = list(self._collect_allowed_items(items_iter, user))
if self._sharing._enabled:
if http_depth == "1":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: get shared collections")
# check for shared collections
collections_shared_map = self._sharing.sharing_collection_map_list(user)
if collections_shared_map:
for sharing in collections_shared_map:
c_share = sharing['PathOrToken']
c_path = sharing['PathMapped']
c_user = sharing['Owner']
c_permissions_filter = sharing['Permissions']
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: test shared collection: PathOrToken=%r PathMapped=%r Owner=%r Permissions=%s", c_share, c_path, c_user, c_permissions_filter)
c_access = Access(self._rights, c_user, c_path, c_permissions_filter)
if not c_access.check("r"):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: skip shared collection: PathOrToken=%r PathMapped=%r Owner=%r Permissions=%s (permissions not matching)", c_share, c_path, c_user, c_permissions_filter)
continue
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: append shared collection: PathOrToken=%r PathMapped=%r Owner=%r", c_share, c_path, c_user)
with self._storage.acquire_lock("r", c_user):
c_items_iter = iter(self._storage.discover(c_path, "0"))
c_allowed_items = list(self._collect_allowed_items(c_items_iter, c_user))
allowed_items = allowed_items + c_allowed_items
headers = {"DAV": httputils.DAV_HEADERS,
"Content-Type": "text/xml; charset=%s" % self._encoding}
xml_answer = xml_propfind(base_prefix, path, xml_content,
allowed_items, user, self._encoding, max_resource_size=self._max_resource_size, sharing=sharing)
if xml_answer is None:
return httputils.NOT_ALLOWED
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)

View File

@@ -4,7 +4,7 @@
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2020 Unrud <unrud@outlook.com>
# Copyright © 2020-2020 Tuna Celik <tuna@jakpark.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-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
@@ -24,7 +24,7 @@ import re
import socket
import xml.etree.ElementTree as ET
from http import client
from typing import Dict, Optional, cast
from typing import Dict, Optional, Union, cast
import defusedxml.ElementTree as DefusedET
@@ -37,7 +37,7 @@ from radicale.log import logger
def xml_proppatch(base_prefix: str, path: str,
xml_request: Optional[ET.Element],
collection: storage.BaseCollection) -> ET.Element:
collection: storage.BaseCollection, sharing: Union[dict, None] = None) -> ET.Element:
"""Read and answer PROPPATCH requests.
Read rfc4918-9.2 for info.
@@ -48,6 +48,9 @@ def xml_proppatch(base_prefix: str, path: str,
multistatus.append(response)
href = ET.Element(xmlutils.make_clark("D:href"))
href.text = xmlutils.make_href(base_prefix, path)
if sharing:
# backmap
href.text = href.text.replace(sharing['PathMapped'], sharing['PathOrToken'])
response.append(href)
# Create D:propstat element for props with status 200 OK
propstat = ET.Element(xmlutils.make_clark("D:propstat"))
@@ -75,7 +78,17 @@ class ApplicationPartProppatch(ApplicationBase):
def do_PROPPATCH(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage PROPPATCH request."""
access = Access(self._rights, user, path)
permissions_filter = None
sharing = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"):
return httputils.NOT_ALLOWED
try:
@@ -99,7 +112,7 @@ class ApplicationPartProppatch(ApplicationBase):
"Content-Type": "text/xml; charset=%s" % self._encoding}
try:
xml_answer = xml_proppatch(base_prefix, path, xml_content,
item)
item, sharing)
if xml_content is not None:
content = DefusedET.tostring(
xml_content,

View File

@@ -4,7 +4,7 @@
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2020 Unrud <unrud@outlook.com>
# Copyright © 2020-2023 Tuna Celik <tuna@jakpark.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -181,7 +181,17 @@ class ApplicationPartPut(ApplicationBase):
def do_PUT(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage PUT request."""
access = Access(self._rights, user, path)
permissions_filter = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"):
return httputils.NOT_ALLOWED
try:

View File

@@ -6,7 +6,7 @@
# Copyright © 2024-2024 Pieter Hijma <pieterhijma@users.noreply.github.com>
# Copyright © 2024-2024 Ray <ray@react0r.com>
# Copyright © 2024-2025 Georgiy <metallerok@gmail.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-2026 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-2025 David Greaves <david@dgreaves.com>
#
# This library is free software: you can redistribute it and/or modify
@@ -25,6 +25,7 @@
import contextlib
import copy
import datetime
import logging
import posixpath
import socket
import xml.etree.ElementTree as ET
@@ -149,14 +150,15 @@ def free_busy_report(base_prefix: str, path: str, xml_request: Optional[ET.Eleme
def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
collection: storage.BaseCollection, encoding: str,
unlock_storage_fn: Callable[[], None],
max_occurrence: int = 0, user: str = "", remote_addr: str = "", remote_useragent: str = ""
) -> Tuple[int, ET.Element]:
max_occurrence: int = 0, user: str = "", remote_addr: str = "", remote_useragent: str = "",
sharing: Union[dict, None] = None) -> Tuple[int, ET.Element]:
"""Read and answer REPORT requests that return XML.
Read rfc3253-3.6 for info.
"""
logger.debug("TRACE/REPORT/xml_report: base_prefix=%r path=%r", base_prefix, path)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: base_prefix=%r path=%r", base_prefix, path)
multistatus = ET.Element(xmlutils.make_clark("D:multistatus"))
if xml_request is None:
return client.MULTI_STATUS, multistatus
@@ -240,7 +242,8 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
filter_copy = copy.deepcopy(filter_)
if expand is not None:
logger.debug("TRACE/REPORT/xml_report: expand")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: expand")
for comp_filter in filter_copy.findall(".//" + xmlutils.make_clark("C:comp-filter")):
if comp_filter.get("name", "").upper() == "VCALENDAR":
continue
@@ -324,13 +327,16 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
n_vevents += n_vev
if prop.tag == xmlutils.make_clark("D:getetag"):
if n_vev > 0:
logger.debug("TRACE/REPORT/xml_report: getetag/expanded element")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: getetag/expanded element")
element.text = item.etag
found_props.append(element)
else:
logger.debug("TRACE/REPORT/xml_report: getetag/no expanded element")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: getetag/no expanded element")
else:
logger.debug("TRACE/REPORT/xml_report: default")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: default")
found_props.append(expanded_element)
else:
if prop.tag == xmlutils.make_clark("D:getetag"):
@@ -354,7 +360,7 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
if found_props or not_found_props:
multistatus.append(xml_item_response(
base_prefix, uri, found_props=found_props,
not_found_props=not_found_props, found_item=True))
not_found_props=not_found_props, found_item=True, sharing=sharing))
return client.MULTI_STATUS, multistatus
@@ -705,11 +711,13 @@ def _find_overridden(
def xml_item_response(base_prefix: str, href: str,
found_props: Sequence[ET.Element] = (),
not_found_props: Sequence[ET.Element] = (),
found_item: bool = True) -> ET.Element:
found_item: bool = True, sharing: Union[dict, None] = None) -> ET.Element:
response = ET.Element(xmlutils.make_clark("D:response"))
href_element = ET.Element(xmlutils.make_clark("D:href"))
href_element.text = xmlutils.make_href(base_prefix, href)
if sharing:
href_element.text = href_element.text.replace(sharing['PathMapped'], sharing['PathOrToken'])
response.append(href_element)
if found_item:
@@ -772,7 +780,8 @@ def retrieve_items(
else:
yield item, False
if collection_requested:
logger.debug("TRACE/REPORT/retrieve_items: get_filtered")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/retrieve_items: get_filtered")
yield from collection.get_filtered(filters)
@@ -810,7 +819,17 @@ class ApplicationPartReport(ApplicationBase):
def do_REPORT(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage REPORT request."""
access = Access(self._rights, user, path)
permissions_filter = None
sharing = None
if self._sharing._enabled:
# Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user)
if sharing:
# overwrite and run through extended permission check
path = sharing['PathMapped']
user = sharing['Owner']
permissions_filter = sharing['Permissions']
access = Access(self._rights, user, path, permissions_filter)
if not access.check("r"):
return httputils.NOT_ALLOWED
try:
@@ -852,7 +871,7 @@ class ApplicationPartReport(ApplicationBase):
try:
status, xml_answer = xml_report(
base_prefix, path, xml_content, collection, self._encoding,
lock_stack.close, max_occurrence, user, remote_host, remote_useragent)
lock_stack.close, max_occurrence, user, remote_host, remote_useragent, sharing=sharing)
except ValueError as e:
logger.warning(
"Bad REPORT request on %r: %s", path, e, exc_info=True)

View File

@@ -34,6 +34,8 @@ class Auth(auth.BaseAuth):
def __init__(self, configuration):
super().__init__(configuration)
self._endpoint = configuration.get("auth", "oauth2_token_endpoint")
self._client_id = configuration.get("auth", "oauth2_client_id")
self._client_secret = configuration.get("auth", "oauth2_client_secret")
if not self._endpoint:
logger.error("auth.oauth2_token_endpoint URL missing")
raise RuntimeError("OAuth2 token endpoint URL is required")
@@ -49,8 +51,11 @@ class Auth(auth.BaseAuth):
"username": login,
"password": password,
"grant_type": "password",
"client_id": "radicale",
"client_id": self._client_id,
}
if self._client_secret:
req_params["client_secret"] = self._client_secret
req_headers = {"Content-Type": "application/x-www-form-urlencoded"}
response = requests.post(
self._endpoint, data=req_params, headers=req_headers

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2017-2020 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -37,7 +37,7 @@ from configparser import RawConfigParser
from typing import (Any, Callable, ClassVar, Iterable, List, Optional,
Sequence, Tuple, TypeVar, Union)
from radicale import auth, hook, rights, storage, types, web
from radicale import auth, hook, rights, sharing, storage, types, web
from radicale.hook import email
from radicale.item import check_and_sanitize_props
@@ -350,6 +350,14 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
"value": "",
"help": "OAuth2 token endpoint URL",
"type": str}),
("oauth2_client_id", {
"value": "radicale",
"help": "OAuth2 client id",
"type": str}),
("oauth2_client_secret", {
"value": "",
"help": "OAuth2 client secret",
"type": str}),
("pam_group_membership", {
"value": "",
"help": "PAM group user should be member of",
@@ -450,6 +458,24 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
"value": "",
"help": "predefined user collections",
"type": json_str})])),
("sharing", OrderedDict([
("type", {
"value": "none",
"help": "sharing database type",
"type": str_or_callable,
"internal": sharing.INTERNAL_TYPES}),
("database_path", {
"value": "",
"help": "database path",
"type": filepath}),
("collection_by_map", {
"value": "false",
"help": "enable sharing of collection by map",
"type": bool}),
("collection_by_token", {
"value": "false",
"help": "enable sharing of collection by token",
"type": bool})])),
("hook", OrderedDict([
("type", {
"value": "none",

View File

@@ -3,7 +3,7 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-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
@@ -106,6 +106,10 @@ MIMETYPES: Mapping[str, str] = {
FALLBACK_MIMETYPE: str = "application/octet-stream"
def bad_request(additional_details: str) -> types.WSGIResponse:
return (client.BAD_REQUEST, (("Content-Type", "text/plain"),), f"Bad Request: {additional_details}", None)
def decode_request(configuration: "config.Configuration",
environ: types.WSGIEnviron, text: bytes) -> str:
"""Try to magically decode ``text`` according to given ``environ``."""

View File

@@ -2,7 +2,7 @@
# Copyright © 2014 Jean-Marc Martins
# Copyright © 2012-2017 Guillaume Ayoub
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-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
@@ -293,6 +293,10 @@ def path_to_filesystem(root: str, sane_path: str) -> str:
return safe_path
def parent_path(path: str) -> str:
return unstrip_path(posixpath.dirname(strip_path(path)), True)
class UnsafePathError(ValueError):
def __init__(self, path: str) -> None:

View File

@@ -0,0 +1,916 @@
# This file is part of Radicale Server - Calendar 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/>.
import base64
import io
import json
import logging
import re
import socket
import uuid
from csv import DictWriter
from datetime import datetime
from http import client
from typing import Sequence, Union
from urllib.parse import parse_qs
from radicale import (config, httputils, pathutils, rights, storage, types,
utils)
from radicale.log import logger
INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none")
DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated')
DB_FIELDS_V1_BOOL: Sequence[str] = ('EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser')
DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
# ShareType: <token|map>
# PathOrToken: <path|token> [PrimaryKey]
# PathMapped: <path>
# Owner: <owner> (creator of database entry)
# User: <user> (user of database entry)
# Permissions: <radicale permission string>
# EnabledByOwner: True|False (share status "invite/grant")
# EnabledByUser: True|False (share status "accept") - check skipped of Owner==User
# HiddenByOwner: True|False (share exposure controlled by owner)
# HiddenByUser: True|False (share exposure controlled by user) - check skipped if Owner==User
# TimestampCreated: <unixtime> (when created)
# TimestampUpdated: <unixtime> (last update)
SHARE_TYPES: Sequence[str] = ('token', 'map', 'all')
SHARE_TYPES_V1: Sequence[str] = ('token', 'map')
# token: share by secret token (does not require authentication)
# map : share by mapping collection of one user to another as virtual
# all : only supported for "list" and "info"
OUTPUT_TYPES: Sequence[str] = ('csv', 'json', 'txt')
API_HOOKS_V1: Sequence[str] = ('list', 'create', 'delete', 'update', 'hide', 'unhide', 'enable', 'disable', 'info')
# list : list sharings (optional filtered)
# create : create share by token or map
# delete : delete share
# update : update share
# hide : hide share (by user or owner)
# unhide : unhide share (by user or owner)
# enable : hide share (by user or owner)
# disable: unhide share (by user or owner)
# info : display support status and permissions
API_SHARE_TOGGLES_V1: Sequence[str] = ('hide', 'unhide', 'enable', 'disable')
TOKEN_PATTERN_V1: str = "(v1/[a-zA-Z0-9_=\\-]{44})"
PATH_PATTERN: str = "([a-zA-Z0-9/.\\-]+)" # TODO: extend or find better source
USER_PATTERN: str = "([a-zA-Z0-9@]+)" # TODO: extend or find better source
def load(configuration: "config.Configuration") -> "BaseSharing":
"""Load the sharing database module chosen in configuration."""
return utils.load_plugin(INTERNAL_TYPES, "sharing", "Sharing", BaseSharing, configuration)
class BaseSharing:
_storage: storage.BaseStorage
_rights: rights.BaseRights
_enabled: bool = False
def __init__(self, configuration: "config.Configuration") -> None:
"""Initialize Sharing.
``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._rights = rights.load(configuration)
self._storage = storage.load(configuration)
# Sharing
self.sharing_collection_by_map = configuration.get("sharing", "collection_by_map")
self.sharing_collection_by_token = configuration.get("sharing", "collection_by_token")
logger.info("sharing.collection_by_map : %s", self.sharing_collection_by_map)
logger.info("sharing.collection_by_token: %s", self.sharing_collection_by_token)
if ((self.sharing_collection_by_map is False) and (self.sharing_collection_by_token is False)):
logger.info("sharing disabled as no feature is enabled")
self._enabled = False
return
else:
self._enabled = True
# database tasks
self.sharing_db_type = configuration.get("sharing", "type")
logger.info("sharing.db_type: %s", self.sharing_db_type)
try:
if self.init_database() is False:
logger.info("sharing disabled as no database is active")
self._enabled = False
return
except Exception as e:
logger.error("sharing database cannot be initialized: %r", e)
exit(1)
database_info = self.get_database_info()
if database_info:
logger.info("sharing database info: %r", database_info)
else:
logger.info("sharing database info: (not provided)")
# overloadable functions
def init_database(self) -> bool:
""" initialize database """
return False
def get_database_info(self) -> Union[dict, None]:
""" retrieve database information """
return None
def verify_database(self) -> bool:
""" verify database information """
return False
def list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
""" retrieve sharing """
return []
def get_sharing(self,
ShareType: str,
PathOrToken: str,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
return {"status": "not-implemented"}
def create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0) -> dict:
""" create sharing """
return {"status": "not-implemented"}
def update_sharing(self,
ShareType: str,
PathOrToken: str,
Owner: Union[str, None] = None,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
Timestamp: int = 0) -> dict:
""" update sharing """
return {"status": "not-implemented"}
def delete_sharing(self,
ShareType: str,
PathOrToken: str,
Owner: str,
PathMapped: Union[str, None] = None) -> dict:
""" delete sharing """
return {"status": "not-implemented"}
def toggle_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
Action: str,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
Timestamp: int = 0) -> dict:
""" toggle sharing """
return {"status": "not-implemented"}
# sharing functions called by request methods
def verify(self) -> bool:
""" verify database """
logger.info("sharing database verification begin")
logger.info("sharing database verification call: %s", self.sharing_db_type)
result = self.verify_database()
if result is not True:
logger.error("sharing database verification call -> PROBLEM: %s", self.sharing_db_type)
return False
else:
pass
logger.info("sharing database verification call -> OK: %s", self.sharing_db_type)
# check all entries
logger.info("sharing database verification content start")
with self._storage.acquire_lock("r"):
for entry in self.list_sharing():
logger.debug("analyze: %r", entry)
if entry['ShareType'] not in SHARE_TYPES_V1:
logger.error("ShareType not supported: %r", entry['ShareType'])
return False
elif not entry['PathMapped'].endswith("/"):
logger.error("PathMapped not ending with '/': %r", entry['PathMapped'])
return False
elif entry['ShareType'] == "map":
if not entry['PathOrToken'].endswith("/"):
logger.error("PathOrToken not ending with '/': %r", entry['PathOrToken'])
return False
else:
pass
# TODO: check PathMapped exists
logger.info("sharing database verification content successful")
return True
def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
if self.sharing_collection_by_token:
result = self.sharing_collection_by_token_resolver(path)
if result is not None:
return result
else:
# check for map
pass
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: not active")
return None
if self.sharing_collection_by_map:
result = self.sharing_collection_by_map_resolver(path, user)
if result is not None:
return result
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
return None
# final
return None
# list sharings of type "map"
def sharing_collection_map_list(self, user: str, active: bool = True) -> list[dict]:
""" returning dict with shared collections (active==True: enabled and unhidden) or None if not found"""
if not self.sharing_collection_by_map:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
return [{}]
# retrieve collections which are enabled and not hidden by owner+user
if active:
shared_collection_list = self.list_sharing(
ShareType="map",
OwnerOrUser=user,
User=user,
EnabledByOwner=True,
EnabledByUser=True,
HiddenByOwner=False,
HiddenByUser=False)
else:
# unconditional
shared_collection_list = self.list_sharing(
ShareType="map",
OwnerOrUser=user,
User=user)
# final
return shared_collection_list
# internal sharing functions
def sharing_collection_by_token_resolver(self, path) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
if self.sharing_collection_by_token:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: check path: %r", path)
if path.startswith("/.token/"):
pattern = re.compile('^/\\.token/' + TOKEN_PATTERN_V1 + '$')
match = pattern.match(path)
if not match:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: unsupported token: %r", path)
return None
else:
# TODO add token validity checks
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: supported token found in path: %r (token=%r)", path, match[1])
return self.get_sharing(
ShareType="token",
PathOrToken=match[1])
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: no supported prefix found in path: %r", path)
return None
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: not active")
return None
def sharing_collection_by_map_resolver(self, path: str, user: str) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
if self.sharing_collection_by_map:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/resolver: check path: %r", path)
result = self.get_sharing(
ShareType="map",
PathOrToken=path,
User=user)
if result:
return result
else:
# fallback to parent path
parent_path = pathutils.parent_path(path)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/resolver: check parent path: %r", parent_path)
result = self.get_sharing(
ShareType="map",
PathOrToken=parent_path,
User=user)
if result:
result['PathMapped'] = path.replace(parent_path, result['PathMapped'])
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/resolver: PathMapped=%r Permissions=%r by parent_path=%r", result['PathMapped'], result['Permissions'], parent_path)
return result
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not found")
return None
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
return None
# POST API
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse:
# Late import to avoid circular dependency in config
from radicale.app.base import Access
"""POST request.
``base_prefix`` is sanitized and never ends with "/".
``path`` is sanitized and always starts with "/.sharing"
``user`` is empty for anonymous users.
Request:
action: (token|map/list
PathOrToken: <path|token> (optional for filter)
action: (token|map)/create
PathMapped: <path> (mandatory)
Permissions: <Permissions> (default: r)
token -> returns <token>
map
PathOrToken: <path> (mandatory)
User: <target_user> (mandatory)
action: (token|map)/(delete|disable|enable|hide|unhide)
PathOrToken: <path|token> (mandatory)
token
map
PathMapped: <path> (mandator)
User: <target_user>
Response: output format depending on ACCEPT header
action: list
by user-owned filtered sharing list in CSV/JSON/TEXT
actions: (other)
Status in JSON/TEXT (TEXT can be parsed by shell)
"""
if not self.sharing_collection_by_map and not self.sharing_collection_by_token:
# API is not enabled
return httputils.NOT_FOUND
if user == "":
# anonymous users are not allowed
return httputils.NOT_ALLOWED
# supported API version check
if not path.startswith("/.sharing/v1/"):
return httputils.NOT_FOUND
# split into ShareType and action or "info"
ShareType_action = path.removeprefix("/.sharing/v1/")
match = re.search('([a-z]+)/([a-z]+)$', ShareType_action)
if not match:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: ShareType/action not extractable: %r", ShareType_action)
return httputils.NOT_FOUND
else:
ShareType = match.group(1)
action = match.group(2)
# check for valid ShareTypes
if ShareType:
if ShareType not in SHARE_TYPES:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: ShareType not whitelisted: %r", ShareType)
return httputils.NOT_FOUND
# check for enabled ShareTypes
if not self.sharing_collection_by_map and ShareType == "map":
# API "map" is not enabled
return httputils.NOT_FOUND
if not self.sharing_collection_by_token and ShareType == "token":
# API "token" is not enabled
return httputils.NOT_FOUND
# check for valid API hooks
if action not in API_HOOKS_V1:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: action not whitelisted: %r", action)
return httputils.NOT_FOUND
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: called by authenticated user: %r", user)
# read POST data
try:
request_body = httputils.read_request_body(self.configuration, environ)
except RuntimeError as e:
logger.warning("Bad POST request on %r (read_request_body): %s", path, e, exc_info=True)
return httputils.bad_request("Failed read POST request body")
except socket.timeout:
logger.debug("Client timed out", exc_info=True)
return httputils.REQUEST_TIMEOUT
api_info = "sharing/API/POST/" + ShareType + "/" + action
# parse body according to content-type
content_type = environ.get("CONTENT_TYPE", "")
if 'application/json' in content_type:
try:
request_data = json.loads(request_body)
except json.JSONDecodeError:
return httputils.bad_request("Invalid JSON")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + " (json): %r", f"{request_data}")
elif 'application/x-www-form-urlencoded' in content_type:
request_parsed = parse_qs(request_body)
# convert arrays into single value
request_data = {}
for key in request_parsed:
request_data[key] = request_parsed[key][0]
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + " (form): %r", f"{request_data}")
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": no supported content data")
return httputils.bad_request("Content-type not supported")
# check for requested output type
accept = environ.get("HTTP_ACCEPT", "")
if 'application/json' in accept:
output_format = "json"
elif 'text/csv' in accept:
output_format = "csv"
else:
output_format = "txt"
if output_format == "csv":
if not action == "list":
return httputils.bad_request("CSV output format is only allowed for list action")
elif output_format == "json":
pass
elif output_format == "txt":
pass
else:
return httputils.bad_request("Output format not supported")
# parameters default
PathOrToken: Union[str, None] = None
PathMapped: Union[str, None] = None
Owner: str = user
User: Union[str, None] = None
Permissions: Union[str, None] = None # no permissions by default
EnabledByOwner: Union[bool, None] = None
HiddenByOwner: Union[bool, None] = None
EnabledByUser: Union[bool, None] = None
HiddenByUser: Union[bool, None] = None
# parameters sanity check
for key in request_data:
if key == "Permissions":
if not re.search('^[a-zA-Z]+$', request_data[key]):
return httputils.bad_request("Invalid value for Permissions")
elif key == "PathOrToken":
if ShareType == "token":
if not re.search('^' + TOKEN_PATTERN_V1 + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken")
elif ShareType == "map":
if not re.search('^' + PATH_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken")
elif not request_data[key].endswith("/"):
return httputils.bad_request("PathOrToken not ending with /")
elif key == "PathMapped":
if not re.search('^' + PATH_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathMapped")
elif not request_data[key].endswith("/"):
return httputils.bad_request("PathMapped not ending with /")
elif key == "Enabled" or key == "Hidden":
if not re.search('^(False|True)$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for " + key)
elif key == "User":
if not re.search('^' + USER_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for User")
# check for mandatory parameters
if 'PathMapped' not in request_data:
if action == 'info':
# ignored
pass
elif action == "list":
# optional
pass
else:
if ShareType == "token" and action != 'create':
# optional
pass
else:
logger.error(api_info + ": missing PathMapped")
return httputils.bad_request("Missing PathMapped")
else:
PathMapped = request_data['PathMapped']
if 'PathOrToken' not in request_data:
if action == 'info':
# ignored
pass
elif action not in ['list', 'create']:
logger.error(api_info + ": missing PathOrToken")
return httputils.bad_request("Missing PathOrToken")
else:
# PathOrToken is optional
pass
else:
if action == "create" and ShareType == "token":
# not supported
logger.error(api_info + ": PathOrToken found but not supported")
return httputils.bad_request("PathOrToken not supported")
PathOrToken = request_data['PathOrToken']
if 'Permissions' in request_data:
Permissions = request_data['Permissions']
if ShareType == "map":
if action == 'info':
# ignored
pass
else:
if 'User' not in request_data:
if action not in ['list', 'delete', 'update']:
logger.warning(api_info + ": missing User")
return httputils.bad_request("Missing User")
else:
# optional
pass
else:
User = request_data['User']
answer: dict = {}
result: dict = {}
result_array: list[dict]
answer['ApiVersion'] = 1
Timestamp = int((datetime.now() - datetime(1970, 1, 1)).total_seconds())
# action: list
if action == "list":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if 'PathOrToken' in request_data:
PathOrToken = request_data['PathOrToken']
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": filter: %r", PathOrToken)
if ShareType != "all":
result_array = self.list_sharing(
ShareType=ShareType,
OwnerOrUser=Owner,
PathMapped=PathMapped,
PathOrToken=PathOrToken)
else:
result_array = self.list_sharing(
OwnerOrUser=Owner,
PathMapped=PathMapped,
PathOrToken=PathOrToken)
answer['Lines'] = len(result_array)
if len(result_array) == 0:
answer['Status'] = "not-found"
else:
answer['Status'] = "success"
answer['Content'] = result_array
# action: create
elif action == "create":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if 'Permissions' not in request_data:
Permissions = "r"
if 'Enabled' in request_data:
EnabledByOwner = config._convert_to_bool(request_data['Enabled'])
else:
EnabledByOwner = False # security by default
if 'Hidden' in request_data:
HiddenByOwner = config._convert_to_bool(request_data['Hidden'])
else:
HiddenByOwner = True # security by default
EnabledByUser = False # security by default
HiddenByUser = True # security by default
if ShareType == "token":
# check access Permissions
access = Access(self._rights, user, str(PathMapped)) # PathMapped is mandatory
if not access.check("r") and "i" not in access.permissions:
logger.info("Add sharing-by-token: access to %r not allowed for user %r", PathMapped, user)
return httputils.NOT_ALLOWED
# v1: create uuid token with 2x 32 bytes = 256 bit
token = "v1/" + str(base64.urlsafe_b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes), 'utf-8')
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": %r (Permissions=%r token=%r)", PathMapped, Permissions, token)
result = self.create_sharing(
ShareType=ShareType,
PathOrToken=token,
PathMapped=str(PathMapped), # mandatory
Owner=Owner, User=Owner,
Permissions=str(Permissions), # mandantory
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
Timestamp=Timestamp)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": result=%r", result)
elif ShareType == "map":
# check preconditions
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
else:
PathOrToken = str(PathOrToken)
if User is None:
return httputils.bad_request("Missing User")
else:
User = str(User)
# check access Permissions
access = Access(self._rights, Owner, str(PathMapped), None) # PathMapped is mandatory
if not access.check("r") and "i" not in access.permissions:
logger.info("Add sharing-by-map: access to path(mapped) %r not allowed for owner %r", PathMapped, Owner)
return httputils.NOT_ALLOWED
access = Access(self._rights, str(User), PathOrToken)
if not access.check("r") and "i" not in access.permissions:
logger.info("Add sharing-by-map: access to path %r not allowed for user %r", PathOrToken, user)
return httputils.NOT_ALLOWED
# check whether share is already existing as real collection
with self._storage.acquire_lock("r", user, path=PathOrToken):
item = next(iter(self._storage.discover(PathOrToken)), None)
if not item:
pass
else:
logger.info("Add sharing-by-map: path %r already exists as real collection for user %r", PathOrToken, user)
return httputils.CONFLICT
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": %r (Permissions=%r PathOrToken=%r user=%r)", PathMapped, Permissions, PathOrToken, User)
result = self.create_sharing(
ShareType=ShareType,
PathOrToken=PathOrToken, # verification above that it is not None
PathMapped=str(PathMapped), # mandatory
Owner=Owner,
User=User, # verification above that it is not None
Permissions=str(Permissions), # mandatory
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
EnabledByUser=EnabledByUser, HiddenByUser=HiddenByUser,
Timestamp=Timestamp)
else:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": result=%r", result)
# result handling
if result['status'] == "conflict":
return httputils.CONFLICT
elif result['status'] == "error":
return httputils.INTERNAL_SERVER_ERROR
elif result['status'] == "success":
answer['Status'] = "success"
else:
return httputils.bad_request("Internal failure")
if ShareType == "token":
logger.info(api_info + "(success): %r (Permissions=%r token=%r)", PathMapped, Permissions, token)
answer['PathOrToken'] = token
# action: update
elif action == "update":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
if ShareType == "token":
result = self.update_sharing(
ShareType=ShareType,
PathMapped=PathMapped,
Permissions=Permissions,
EnabledByOwner=EnabledByOwner,
HiddenByOwner=HiddenByOwner,
PathOrToken=str(PathOrToken), # verification above that it is not None
Owner=Owner,
Timestamp=Timestamp)
elif ShareType == "map":
result = self.update_sharing(
ShareType=ShareType,
PathMapped=PathMapped,
Permissions=Permissions,
EnabledByOwner=EnabledByOwner,
HiddenByOwner=HiddenByOwner,
PathOrToken=str(PathOrToken), # verification above that it is not None
Owner=Owner,
Timestamp=Timestamp)
else:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
# result handling
if result['status'] == "not-found":
return httputils.NOT_FOUND
elif result['status'] == "permission-denied":
return httputils.NOT_ALLOWED
elif result['status'] == "success":
answer['Status'] = "success"
pass
else:
if ShareType == "token":
logger.info("Update of sharing-by-token: %r not successful", request_data['PathOrToken'])
elif ShareType == "map":
logger.info("Update of sharing-by-map: %r not successful", request_data['PathOrToken'])
return httputils.bad_request("Invalid share type")
# action: delete
elif action == "delete":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
if ShareType == "token":
result = self.delete_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
Owner=Owner)
elif ShareType == "map":
result = self.delete_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
PathMapped=PathMapped,
Owner=Owner)
else:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
# result handling
if result['status'] == "not-found":
return httputils.NOT_FOUND
elif result['status'] == "permission-denied":
return httputils.NOT_ALLOWED
elif result['status'] == "success":
answer['Status'] = "success"
pass
else:
if ShareType == "token":
logger.info("Delete sharing-by-token: %r of user %r not successful", request_data['PathOrToken'], request_data['User'])
elif ShareType == "map":
logger.info("Delete sharing-by-map: %r of user %r not successful", request_data['PathOrToken'], request_data['User'])
return httputils.bad_request("Invalid share type")
# action: info
elif action == "info":
answer['Status'] = "success"
if ShareType in ["all", "map"]:
answer['FeatureEnabledCollectionByMap'] = self.sharing_collection_by_map
answer['PermittedCreateCollectionByMap'] = True # TODO toggle per permission, default?
if ShareType in ["all", "token"]:
answer['FeatureEnabledCollectionByToken'] = self.sharing_collection_by_token
answer['PermittedCreateCollectionByToken'] = True # TODO toggle per permission, default?
# action: TOGGLE
elif action in API_SHARE_TOGGLES_V1:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/POST/" + action)
if ShareType in ["token", "map"]:
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
result = self.toggle_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
OwnerOrUser=user, # authenticated user
User=User, # optional for selection
PathMapped=PathMapped, # optional for selection
Action=action,
Timestamp=Timestamp)
if result:
if result['status'] == "not-found":
return httputils.NOT_FOUND
if result['status'] == "permission-denied":
return httputils.NOT_ALLOWED
elif result['status'] == "success":
answer['Status'] = "success"
pass
else:
logger.error("Toggle sharing: %r of user %s not successful", request_data['PathOrToken'], user)
return httputils.bad_request("Internal Error")
else:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
else:
# default
logger.error(api_info + ": unsupported action=%r", action)
return httputils.bad_request("Invalid action")
# output handler
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/POST output format: %r", output_format)
logger.debug("TRACE/sharing/API/POST answer: %r", answer)
if output_format == "csv" or output_format == "txt":
answer_array = []
if output_format == "txt":
for key in answer:
if key != 'Content':
answer_array.append(key + '=' + str(answer[key]))
if 'Content' in answer and answer['Content'] is not None:
csv = io.StringIO()
writer = DictWriter(csv, fieldnames=DB_FIELDS_V1)
if output_format == "csv":
writer.writeheader()
for entry in answer['Content']:
writer.writerow(entry)
if output_format == "csv":
answer_array.append(csv.getvalue())
else:
index = 0
for line in csv.getvalue().splitlines():
# create a shell array with content lines
answer_array.append('Content[' + str(index) + ']="' + line + '"')
index += 1
headers = {
"Content-Type": "text/csv"
}
return client.OK, headers, "\n".join(answer_array), None
elif output_format == "json":
answer_raw = json.dumps(answer)
headers = {
"Content-Type": "text/json"
}
return client.OK, headers, answer_raw, None
else:
# should not be reached
return httputils.bad_request("Invalid output format")
return httputils.METHOD_NOT_ALLOWED

530
radicale/sharing/csv.py Normal file
View File

@@ -0,0 +1,530 @@
# This file is part of Radicale Server - Calendar 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/>.
import csv
import logging
import os
from typing import Union
from radicale import config, sharing
from radicale.log import logger
""" CVS based sharing by token or map """
class Sharing(sharing.BaseSharing):
_lines: int = 0
_sharing_cache: list[dict] = []
_sharing_db_file: str
# Overloaded functions
def init_database(self) -> bool:
logger.debug("sharing database initialization for type 'csv'")
sharing_db_file = self.configuration.get("sharing", "database_path")
if sharing_db_file == "":
folder = self.configuration.get("storage", "filesystem_folder")
folder_db = os.path.join(folder, "collection-db")
sharing_db_file = os.path.join(folder_db, "sharing.csv")
logger.info("sharing database filename not provided, use default: %r", sharing_db_file)
else:
logger.info("sharing database filename: %r", sharing_db_file)
if not os.path.exists(folder_db):
logger.warning("sharing database folder is not existing: %r (create now)", folder_db)
try:
os.mkdir(folder_db)
except Exception as e:
logger.error("sharing database folder cannot be created (check permissions): %r (%r)", folder_db, e)
return False
logger.info("sharing database folder successfully created: %r", folder_db)
if not os.path.exists(sharing_db_file):
logger.warning("sharing database is not existing: %r", sharing_db_file)
try:
if self._create_empty_csv(sharing_db_file) is not True:
raise
except Exception as e:
logger.error("sharing database (empty) cannot be created (check permissions): %r (%r)", sharing_db_file, e)
return False
logger.info("sharing database (empty) successfully created: %r", sharing_db_file)
else:
logger.info("sharing database exists: %r", sharing_db_file)
# read database
try:
if self._load_csv(sharing_db_file) is not True:
return False
except Exception as e:
logger.error("sharing database load failed: %r (%r)", sharing_db_file, e)
return False
logger.info("sharing database load successful: %r (lines=%d)", sharing_db_file, self._lines)
self._sharing_db_file = sharing_db_file
return True
def get_database_info(self) -> Union[dict, None]:
database_info = {'type': "csv"}
return database_info
def verify_database(self) -> bool:
logger.info("sharing database (csv) verification begin")
logger.info("sharing database (csv) file: %r", self._sharing_db_file)
logger.info("sharing database (csv) loaded entries: %d", self._lines)
# nothing more todo for CSV
logger.info("sharing database (csv) verification end")
return True
def get_sharing(self,
ShareType: str,
PathOrToken: str,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
# Lookup
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: lookup ShareType=%r PathOrToken=%r User=%r)", ShareType, PathOrToken, User)
index = 0
found = False
for row in self._sharing_cache:
if index == 0:
# skip fieldnames
pass
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: check row: %r", row)
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
elif User is not None and row['User'] != User:
pass
elif row['EnabledByOwner'] is not True:
pass
elif row['ShareType'] == "map":
if row['EnabledByUser'] is not True:
pass
else:
found = True
break
else:
found = True
break
index += 1
if found:
PathMapped = row['PathMapped']
Owner = row['Owner']
UserShare = row['User']
Permissions = row['Permissions']
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden)
return {
"mapped": True,
"PathOrToken": PathOrToken,
"PathMapped": PathMapped,
"Owner": Owner,
"User": UserShare,
"Hidden": Hidden,
"Permissions": Permissions}
return None
def list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
""" retrieve sharing """
row: dict
index = 0
result = []
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r HiddenByOwner=%s HiddenByUser=%s", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, HiddenByOwner, HiddenByUser)
for row in self._sharing_cache:
if index == 0:
# skip fieldnames
pass
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: test: %r", row)
if ShareType is not None and row['ShareType'] != ShareType:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("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:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: skip by User")
pass
elif PathOrToken is not None and row['PathOrToken'] != PathOrToken:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: skip by PathOrToken")
pass
elif PathMapped is not None and row['PathMapped'] != PathMapped:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("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
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: add: %r", row)
result.append(row)
index += 1
return result
def create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0) -> dict:
""" create sharing """
row: dict
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: ShareType=%r", ShareType)
if ShareType == "token":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", PathOrToken, Owner, PathMapped, User, Permissions)
# check for duplicate token entry
for row in self._sharing_cache:
if row['ShareType'] != "token":
continue
if row['PathOrToken'] == PathOrToken:
# must be unique systemwide
logger.error("sharing/token/create: PathOrToken already exists: PathOrToken=%r", PathOrToken)
return {"status": "conflict"}
elif ShareType == "map":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", PathOrToken, Owner, PathMapped, User, Permissions)
# check for duplicate map entry
for row in self._sharing_cache:
if row['ShareType'] != "map":
continue
if row['PathMapped'] == PathMapped and row['User'] == User and row['PathOrToken'] == PathOrToken:
# must be unique systemwide
logger.error("sharing/map/create: entry already exists: PathMapped=%r User=%r", PathMapped, User)
return {"status": "conflict"}
else:
return {"status": "error"}
row = {"ShareType": ShareType,
"PathOrToken": PathOrToken,
"PathMapped": PathMapped,
"Owner": Owner,
"User": User,
"Permissions": Permissions,
"EnabledByOwner": EnabledByOwner,
"EnabledByUser": EnabledByUser,
"HiddenByOwner": HiddenByOwner,
"HiddenByUser": HiddenByUser,
"TimestampCreated": Timestamp,
"TimestampUpdated": Timestamp}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/*/create: add row: %r", row)
self._sharing_cache.append(row)
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: write CSV done", ShareType)
return {"status": "success"}
logger.error("sharing/%s/create: cannot update CSV database", ShareType)
return {"status": "error"}
def update_sharing(self,
ShareType: str,
PathOrToken: str,
Owner: Union[str, None] = None,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
Timestamp: int = 0) -> dict:
""" update sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped)
# lookup token
found = False
index = 0
for row in self._sharing_cache:
if index == 0:
# skip fieldnames
pass
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
else:
found = True
break
index += 1
if found:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index)
if Owner is not None and row['Owner'] != Owner:
return {"status": "permission-denied"}
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index)
logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row)
# CSV: remove+adjust+readd
if PathMapped is not None:
row["PathMapped"] = PathMapped
if Permissions is not None:
row["Permissions"] = Permissions
if User is not None:
row["User"] = User
if EnabledByOwner is not None:
row["EnabledByOwner"] = EnabledByOwner
if HiddenByOwner is not None:
row["HiddenByOwner"] = HiddenByOwner
# update timestamp
row["TimestampUpdated"] = Timestamp
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: adj row=%r", ShareType, row)
# replace row
self._sharing_cache.pop(index)
self._sharing_cache.append(row)
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: write CSV done", ShareType)
return {"status": "success"}
logger.error("sharing/%s/update: cannot update CSV database", ShareType)
return {"status": "error"}
else:
return {"status": "not-found"}
def delete_sharing(self,
ShareType: str,
PathOrToken: str, Owner: str,
PathMapped: Union[str, None] = None) -> dict:
""" delete sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped)
# lookup token
found = False
index = 0
for row in self._sharing_cache:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: check: %r", ShareType, row)
if index == 0:
# skip fieldnames
pass
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
else:
if ShareType == "map":
# extra filter
if row['PathMapped'] != PathMapped:
pass
else:
found = True
break
else:
found = True
break
index += 1
if found:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: found index=%d", ShareType, index)
if row['Owner'] != Owner:
return {"status": "permission-denied"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index)
self._sharing_cache.pop(index)
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing_by_token: write CSV done")
return {"status": "success"}
logger.error("sharing/%s/delete: cannot update CSV database", ShareType)
return {"status": "error"}
else:
return {"status": "not-found"}
def toggle_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
Action: str,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
Timestamp: int = 0) -> dict:
""" toggle sharing """
row: dict
if Action not in sharing.API_SHARE_TOGGLES_V1:
# should not happen
raise
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r", ShareType, Action, OwnerOrUser, User, PathOrToken, PathMapped)
# lookup entry
found = False
index = 0
for row in self._sharing_cache:
if index == 0:
# skip fieldnames
pass
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/*/" + Action + ": check: %r", row)
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
elif PathMapped is not None and row['PathMapped'] != PathMapped:
pass
elif row['Owner'] == OwnerOrUser:
found = True
break
else:
found = True
break
index += 1
if found:
# if logger.isEnabledFor(logging.DEBUG):
# logger.debug("TRACE/sharing/*/" + Action + ": found: %r", row)
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
elif row['Owner'] == OwnerOrUser:
pass
elif row['User'] == OwnerOrUser:
pass
else:
return {"status": "permission-denied"}
# TODO: locking
if row['Owner'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: Owner=%r User=%r PathOrToken=%r index=%d", ShareType, Action, OwnerOrUser, User, PathOrToken, index)
if Action == "disable":
row['EnabledByOwner'] = False
elif Action == "enable":
row['EnabledByOwner'] = True
elif Action == "hide":
row['HiddenByOwner'] = True
elif Action == "unhide":
row['HiddenByOwner'] = False
row['TimestampUpdated'] = Timestamp
if row['User'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: User=%r PathOrToken=%r index=%d", ShareType, Action, OwnerOrUser, PathOrToken, index)
if Action == "disable":
row['EnabledByUser'] = False
elif Action == "enable":
row['EnabledByUser'] = True
elif Action == "hide":
row['HiddenByUser'] = True
elif Action == "unhide":
row['HiddenByUser'] = False
row['TimestampUpdated'] = Timestamp
# remove
self._sharing_cache.pop(index)
# readd
self._sharing_cache.append(row)
with self._storage.acquire_lock("w", OwnerOrUser, path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE: write CSV done")
return {"status": "success"}
logger.error("sharing: cannot update CSV database")
return {"status": "error"}
else:
return {"status": "not-found"}
# local functions
def _create_empty_csv(self, file: str) -> bool:
with self._storage.acquire_lock("w", None, path=file):
with open(file, 'w', newline='') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1)
writer.writeheader()
return True
def _load_csv(self, file: str) -> bool:
logger.debug("sharing database load begin: %r", file)
with self._storage.acquire_lock("r", None):
with open(file, 'r', newline='') as csvfile:
reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1)
self._lines = 0
for row in reader:
# logger.debug("sharing database load read: %r", row)
if self._lines == 0:
# header line, check
for fieldname in sharing.DB_FIELDS_V1:
logger.debug("sharing database load check fieldname: %r", fieldname)
if fieldname not in row:
logger.debug("sharing database is incompatible: %r", file)
return False
# convert txt to bool
if self._lines > 0:
for fieldname in sharing.DB_FIELDS_V1_BOOL:
row[fieldname] = config._convert_to_bool(row[fieldname])
for fieldname in sharing.DB_FIELDS_V1_INT:
row[fieldname] = int(row[fieldname])
# check for duplicates
dup = False
for row_cached in self._sharing_cache:
if row == row_cached:
dup = True
break
if dup:
continue
# logger.debug("sharing database load add: %r", row)
self._sharing_cache.append(row)
self._lines += 1
logger.debug("sharing database load end: %r", file)
return True
def _write_csv(self, file: str) -> bool:
with open(file, 'w', newline='') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1)
writer.writerows(self._sharing_cache)
return True

445
radicale/sharing/files.py Normal file
View File

@@ -0,0 +1,445 @@
# This file is part of Radicale Server - Calendar 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/>.
import logging
import os
import pickle
import urllib
from typing import Union
from radicale import sharing
from radicale.log import logger
""" File 'database' based sharing by token or map """
DB_VERSION: str = "1"
class Sharing(sharing.BaseSharing):
_sharing_db_path_ShareType: dict = {}
# Overloaded functions
def init_database(self) -> bool:
logger.debug("sharing database initialization for type 'files'")
sharing_db_path = self.configuration.get("sharing", "database_path")
if sharing_db_path == "":
folder = self.configuration.get("storage", "filesystem_folder")
folder_db = os.path.join(folder, "collection-db")
sharing_db_path = os.path.join(folder_db, "files")
logger.info("sharing database path not provided, use default: %r", sharing_db_path)
else:
logger.info("sharing database path: %r", sharing_db_path)
if not os.path.exists(folder_db):
logger.warning("sharing database folder is not existing: %r (create now)", folder_db)
try:
os.mkdir(folder_db)
except Exception as e:
logger.error("sharing database folder cannot be created (check permissions): %r (%r)", folder_db, e)
return False
logger.info("sharing database folder successfully created: %r", folder_db)
if not os.path.exists(sharing_db_path):
logger.warning("sharing database path is not existing: %r", sharing_db_path)
try:
os.mkdir(sharing_db_path)
except Exception as e:
logger.error("sharing database path cannot be created (check permissions): %r (%r)", sharing_db_path, e)
return False
logger.info("sharing database path successfully created: %r", sharing_db_path)
for ShareType in sharing.SHARE_TYPES_V1:
path = os.path.join(sharing_db_path, ShareType)
self._sharing_db_path_ShareType[ShareType] = path
if not os.path.exists(path):
logger.warning("sharing database path for %r is not existing: %r", ShareType, path)
try:
os.mkdir(path)
except Exception as e:
logger.error("sharing database path for %r cannot be created (check permissions): %r (%r)", ShareType, path, e)
return False
logger.info("sharing database path for %r successfully created: %r", ShareType, path)
return True
def get_database_info(self) -> Union[dict, None]:
database_info = {'type': "files"}
return database_info
def verify_database(self) -> bool:
logger.info("sharing database (files) verification begin")
for ShareType in sharing.SHARE_TYPES_V1:
logger.info("sharing database (files) path for %r: %r", ShareType, self._sharing_db_path_ShareType[ShareType])
# TODO: count amount of files
logger.info("sharing database (files) verification end")
return True
def get_sharing(self,
ShareType: str,
PathOrToken: str,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
# Lookup
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/get: PathOrToken=%r User=%r)", ShareType, PathOrToken, User)
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return None
# read content
with self._storage.acquire_lock("r", User):
# read file
with open(sharing_config_file, "rb") as fb:
(version, row) = pickle.load(fb)
if version != DB_VERSION:
return {"status": "error"}
if User is not None and row['User'] != User:
return None
elif row['EnabledByOwner'] is not True:
return None
elif row['ShareType'] == "map":
if row['EnabledByUser'] is not True:
return None
PathMapped = row['PathMapped']
Owner = row['Owner']
UserShare = row['User']
Permissions = row['Permissions']
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden)
return {
"mapped": True,
"PathOrToken": PathOrToken,
"PathMapped": PathMapped,
"Owner": Owner,
"User": UserShare,
"Hidden": Hidden,
"Permissions": Permissions}
return None
def list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
""" retrieve sharing """
result = []
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r HiddenByOwner=%s HiddenByUser=%s", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, HiddenByOwner, HiddenByUser)
for _ShareType in sharing.SHARE_TYPES_V1:
if ShareType is not None and _ShareType != ShareType:
# skip
continue
path = self._sharing_db_path_ShareType[_ShareType]
with self._storage.acquire_lock("r", OwnerOrUser, path=path):
for entry in os.scandir(path):
if not entry.is_file():
continue
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list: check file: %r", entry.name)
# read file
with open(entry, "rb") as fb:
(version, row) = pickle.load(fb)
if version != DB_VERSION:
# skip
continue
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: test: %r", row)
if ShareType is not None and row['ShareType'] != ShareType:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("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:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: skip by User")
pass
elif PathOrToken is not None and row['PathOrToken'] != PathOrToken:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: skip by PathOrToken")
pass
elif PathMapped is not None and row['PathMapped'] != PathMapped:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("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
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: add: %r", row)
result.append(row)
return result
def create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0) -> dict:
""" create sharing """
row: dict
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: sharing_config_file=%r", ShareType, sharing_config_file)
logger.debug("TRACE/sharing/%s/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", ShareType, PathOrToken, Owner, PathMapped, User, Permissions)
if os.path.isfile(sharing_config_file):
return {"status": "conflict"}
row = {"ShareType": ShareType,
"PathOrToken": PathOrToken,
"PathMapped": PathMapped,
"Owner": Owner,
"User": User,
"Permissions": Permissions,
"EnabledByOwner": EnabledByOwner,
"EnabledByUser": EnabledByUser,
"HiddenByOwner": HiddenByOwner,
"HiddenByUser": HiddenByUser,
"TimestampCreated": Timestamp,
"TimestampUpdated": Timestamp}
version = DB_VERSION
try:
with self._storage.acquire_lock("w", Owner, path=sharing_config_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: store share-config: %r into file %r", ShareType, row, sharing_config_file)
# write file
with open(sharing_config_file, "wb") as fb:
pickle.dump((version, row), fb)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/*/create: share-config file stored: %r", sharing_config_file)
return {"status": "success"}
except Exception as e:
logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
def update_sharing(self,
ShareType: str,
PathOrToken: str,
Owner: Union[str, None] = None,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
Timestamp: int = 0) -> dict:
""" update sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r User=%r", ShareType, PathOrToken, Owner, User)
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return {"status": "not-found"}
# read content
with self._storage.acquire_lock("w", Owner, path=sharing_config_file):
# read file
with open(sharing_config_file, "rb") as fb:
(version, row) = pickle.load(fb)
if version != DB_VERSION:
return {"status": "error"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: check: %r", ShareType, row)
if Owner is not None and row['Owner'] != Owner:
return {"status": "permission-denied"}
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row)
if PathMapped is not None:
row["PathMapped"] = PathMapped
if Permissions is not None:
row["Permissions"] = Permissions
if User is not None:
row["User"] = User
if EnabledByOwner is not None:
row["EnabledByOwner"] = EnabledByOwner
if HiddenByOwner is not None:
row["HiddenByOwner"] = HiddenByOwner
# update timestamp
row["TimestampUpdated"] = Timestamp
logger.debug("TRACE/sharing/%s/update: adj row=%r", ShareType, row)
try:
# write file
with open(sharing_config_file, "wb") as fb:
pickle.dump((version, row), fb)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: share-config file stored: %r", ShareType, sharing_config_file)
return {"status": "success"}
except Exception as e:
logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
def delete_sharing(self,
ShareType: str,
PathOrToken: str, Owner: str,
PathMapped: Union[str, None] = None) -> dict:
""" delete sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r", ShareType, PathOrToken, Owner)
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return {"status": "not-found"}
# read content
with self._storage.acquire_lock("r", Owner, path=sharing_config_file):
# read file
with open(sharing_config_file, "rb") as fb:
(version, row) = pickle.load(fb)
if version != DB_VERSION:
return {"status": "error"}
# verify owner
if row['Owner'] != Owner:
return {"status": "permission-denied"}
try:
os.remove(sharing_config_file)
except Exception as e:
logger.error("sharing/%s/delete: cannot remove share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
logger.debug("sharing/%s/delete: successful removed share-config: %r", ShareType, sharing_config_file)
return {"status": "success"}
def toggle_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
Action: str,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
Timestamp: int = 0) -> dict:
""" toggle sharing """
row: dict
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r", ShareType, Action, OwnerOrUser, User, PathOrToken, PathMapped)
if Action not in sharing.API_SHARE_TOGGLES_V1:
# should not happen
raise
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return {"status": "not-found"}
# read content
with self._storage.acquire_lock("w", OwnerOrUser, path=sharing_config_file):
# read file
with open(sharing_config_file, "rb") as fb:
(version, row) = pickle.load(fb)
if version != DB_VERSION:
return {"status": "error"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: check: %r", ShareType, Action, row)
# verify ownership or user
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
elif row['Owner'] == OwnerOrUser:
pass
elif row['User'] == OwnerOrUser:
pass
else:
return {"status": "permission-denied"}
if row['Owner'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: Owner=%r User=%r PathOrToken=%r", ShareType, Action, OwnerOrUser, User, PathOrToken)
if Action == "disable":
row['EnabledByOwner'] = False
elif Action == "enable":
row['EnabledByOwner'] = True
elif Action == "hide":
row['HiddenByOwner'] = True
elif Action == "unhide":
row['HiddenByOwner'] = False
row['TimestampUpdated'] = Timestamp
if row['User'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: User=%r PathOrToken=%r", ShareType, Action, OwnerOrUser, PathOrToken)
if Action == "disable":
row['EnabledByUser'] = False
elif Action == "enable":
row['EnabledByUser'] = True
elif Action == "hide":
row['HiddenByUser'] = True
elif Action == "unhide":
row['HiddenByUser'] = False
row['TimestampUpdated'] = Timestamp
try:
# write file
with open(sharing_config_file, "wb") as fb:
pickle.dump((version, row), fb)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: share-config file stored: %r", ShareType, sharing_config_file)
return {"status": "success"}
except Exception as e:
logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
# local functions
def _encode_path(self, path: str) -> str:
return urllib.parse.quote(path, safe="")
def _decode_path(self, path: str) -> str:
return urllib.parse.unquote(path)

38
radicale/sharing/none.py Normal file
View File

@@ -0,0 +1,38 @@
# This file is part of Radicale Server - Calendar 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/>.
from typing import Union
from radicale import sharing
from radicale.log import logger
class Sharing(sharing.BaseSharing):
def init_database(self) -> bool:
""" dummy initialization """
return False
def get_sharing_collection_by_token(self, token: str) -> Union[dict, None]:
""" retrieve target and attributs by token """
# default
logger.debug("TRACE/sharing_by_token: 'none' cannot provide any map for token: %r", token)
return None
def get_sharing_collection_by_map(self, path) -> Union[dict, None]:
""" retrieve target and attributs by map """
logger.debug("TRACE/sharing_by_map: 'none' cannot provide any map for path: %r", path)
return {"mapped": False}

View File

@@ -89,6 +89,8 @@ class BaseTest:
(str, type(http_if_match)))
remote_useragent = kwargs.pop("remote_useragent", None)
remote_host = kwargs.pop("remote_host", None)
content_type = kwargs.pop("content_type", None)
accept = kwargs.pop("accept", None)
environ: Dict[str, Any] = {k.upper(): v for k, v in kwargs.items()}
for k, v in environ.items():
if not isinstance(v, str):
@@ -104,6 +106,10 @@ class BaseTest:
environ["HTTP_USER_AGENT"] = remote_useragent
if remote_host:
environ["REMOTE_ADDR"] = remote_host
if content_type:
environ["CONTENT_TYPE"] = content_type
if accept:
environ["HTTP_ACCEPT"] = accept
environ["REQUEST_METHOD"] = method.upper()
environ["PATH_INFO"] = path
if data is not None:

File diff suppressed because it is too large Load Diff

View File

@@ -20,7 +20,7 @@ from setuptools import find_packages, setup
# When the version is updated, a new section in the CHANGELOG.md file must be
# added too.
VERSION = "3.6.1.dev"
VERSION = "3.7.0.dev"
with open("README.md", encoding="utf-8") as f:
long_description = f.read()