Avii/Radicale
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge pull request #2040 from pbiering/sharing-replace-bday-by-conversion

Browse Source 
Sharing replace bday by conversion and next review
This commit is contained in:
Peter Bieringer
2026-03-25 07:17:02 +01:00
committed by GitHub
parent 980c635665 a923ccf2d3
commit fd067a6a59
13 changed files with 689 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
SHARING.md
View File

@@ -30,12 +30,12 @@ Types of supported sharing configuration:
* `ShareType`: type of share
* `token`: token-based share (do not require user authentication)
* `map`: map-based share (requires user authentication)
* `bday`: map-based share (requires user authentication) with on-the-fly auto-conversion
* `PathOrToken`: token or "virtual" collection, has to be unique (PRIMARY KEY)
* `PathMapped`: target collection
* `Conversion`: conversion method
* `Owner`: owner of the share
* `User`: user of the share
* `Permissions`: effective permission of the share (*bday* is always read-only)
* `Permissions`: effective permission of the share
* `EnabledByOwner`: control by owner
* `EnabledByUser`: control by user
* `HiddenByOwner`: control by owner
@@ -43,11 +43,18 @@ Types of supported sharing configuration:
* `TimestampCreated`: unixtime of creation
* `TimestampUpdated`: unixtime of last update
* `Properties`: overlay properties (limited set whitelisted)
* `Actions`: (reserved for future usage)
`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
#### Supported Conversions
* `none`: no conversion
* `bday`: auto-mapping on-the-fly a VADDRESSBOOK to a VCALENDAR of all entries containing a `BDAY`
* Permissions enforced to read-only
### Sharing Configuration Entry Storage
#### CSV
@@ -168,14 +175,11 @@ File-based configuration store is using encoded `PathOrToken` as filename for ea
Map-based sharing can be accessed as usual after authentication and authorization.
* *map* is a standard sharing of one collection to another user
* *bday* is special sharing auto-mapping on-the-fly a VADDRESSBOOK to a VCALENDAR of all entries containing a `BDAY`
#### Permission Control
* `permit_create_map`
* supported *rights* permissions: `Mm`
* `permit_create_bday`
* supported *rights* permissions: `Bb`
#### Workflow
@@ -264,7 +268,8 @@ Status='success'
FeatureEnabledCollectionByMap=True
PermittedCreateCollectionByMap=True
FeatureEnabledCollectionByToken=True
PermittedCreateCollectionByToken=True
PermittedCreateCollectionByToken=True
SupportedConversions=(bday none)
```
* json->json, parsed with `jq`
@@ -279,12 +284,11 @@ curl -u user:$userpw --silent -H "accept: application/json" -d "" http://localho
"PermittedCreateCollectionByMap": true,
"FeatureEnabledCollectionByToken": true,
"PermittedCreateCollectionByToken": true,
"FeatureEnabledCollectionByBday": true,
"PermittedCreateCollectionByBday": true
"SupportedConversions": ["bday", "none"]
}
```
##### API Hook "(token|map|bday)/create"
##### API Hook "(token|map)/create"
* Authorization
* Authenticated user is `Owner`
@@ -304,6 +308,7 @@ Create a share by mapping a collection of an `Owner` to a token.
| Parameter | Type | Requirement |
| - | - | - |
| PathMapped | str | mandatory |
| Conversion | str | optional(default:none) |
| User | str | optional(default:owner) |
| Permissions | str | optional(default:r) |
| Enabled | bool | optional(owner/default:False) |
@@ -323,17 +328,17 @@ Create a share by mapping a collection of an `Owner` to a token.
curl -u user:$userpw -d "PathMapped=/user/testcalendar1/" -d "Enabled=True" -d "Hidden=False" http://localhost:5232/.sharing/v1/token/create
ApiVersion=1
Status='success'
PathOrToken='v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw='
PathOrToken='/.token/v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw0/'
```
* json->json
```bash
curl -u user:$userpw -H "Content-Type: application/json" -d '{ "PathMapped": "/user/testcalendar1/", "Enabled": true, "Hidden": false}' http://localhost:5232/.sharing/v1/token/create
{"ApiVersion": 1, "Status": "success", "PathOrToken": "v1/aMsmGqOsRwSH-2-6tEa8EMr4RMYzMU7WvPmjnp5qDnw="}
{"ApiVersion": 1, "Status": "success", "PathOrToken": "/.token/v1/aMsmGqOsRwSH-2-6tEa8EMr4RMYzMU7WvPmjnp5qDnw0/"}
```
###### API Hook "(map|bday)/create"
###### API Hook "map/create"
Create a share by mapping a collection of an `Owner` to an `User`.
@@ -342,13 +347,8 @@ Create a share by mapping a collection of an `Owner` to an `User`.
* `PathMapped` is not existing already as a share target for same `User`
* Authenticated user as `Owner` has at least read access to `PathMapped`
* Provided `User` has at least read access to `PathOrToken`
* *map*
* Global permitted by `permit_create_map = True` or `rights` permission `m`
* Global denied by `permit_create_map = False` or `rights` permission `M`
* *bday*
* Global permitted by `permit_create_map = True` or `rights` permission `b`
* Global denied by `permit_create_map = False` or `rights` permission `B`
* `PathMapped` is a VCALENDAR collection
* Global permitted by `permit_create_map = True` or `rights` permission `m`
* Global denied by `permit_create_map = False` or `rights` permission `M`
* Input
@@ -356,6 +356,7 @@ Create a share by mapping a collection of an `Owner` to an `User`.
| - | - |
| PathOrToken | str | mandatory |
| PathMapped | str | mandatory |
| Conversion | str | optional(default:none) |
| User | str | mandatory |
| Permissions | str | optional(default:r) |
| Enabled | bool | optional(owner/default:False) |
@@ -380,7 +381,7 @@ curl -u owner:$ownerpw -H "Content-Type: application/json" -d '{ "PathOrToken":
{"ApiVersion": 1, "Status": "success"}
```
##### API Hook "(all|token|map|bday)/list"
##### API Hook "(all|token|map)/list"
List shares (optional with filter) either owned or assigned as user.
@@ -461,7 +462,7 @@ curl -s -H "Content-Type: application/json" -u user:$userpw -d "{}" http://local
```
##### API Hook "(token|map|bday)/delete"
##### API Hook "(token|map)/delete"
Delete a share selected by `PathOrToken`.
@@ -494,7 +495,7 @@ curl -u user:$userpw -H "Content-Type: application/json" -d '{ "PathOrToken": "v
{"ApiVersion": 1, "Status": "success"}
```
##### API Hook "(token|map|bday)/update"
##### API Hook "(token|map)/update"
Update a share selected by `PathOrToken`.
@@ -534,7 +535,7 @@ curl -u user:$userpw -H "Content-Type: application/json" -d '{ "PathOrToken": "/
{"ApiVersion": 1, "Status": "success"}
```
##### API Hooks "(token|map|bday)/(enable|disable|hide|unhide)"
##### API Hooks "(token|map)/(enable|disable|hide|unhide)"
Toggle enable|disable|hide|unhide of `Owner` or `User` of a share selected by `PathOrToken`
@@ -659,17 +660,16 @@ Preconditions:
* Collection with type *adressbook* is existing
* Config options enabled in section `sharing`:
* `collection_by_bday`
* `permit_create_bday`
* `collection_by_map`
* `permit_create_map`
#### Examples using API
* Create
* Create as *map*
```bash
## Create sharing of type "bday"
curl -u owner:$ownerpw -d "PathOrToken=/owner/bday-of-addressbook/" -d "PathMapped=/owner/addressbook/" -d "User=owner" http://localhost:5232/.sharing/v1/bday/create
## Create sharing of type *map* with conversion *bday*
curl -u owner:$ownerpw -d "PathOrToken=/owner/bday-of-addressbook/" -d "PathMapped=/owner/addressbook/" -d "User=owner" -d "Conversion=bday" http://localhost:5232/.sharing/v1/map/create
## Enable
curl -u owner:$ownerpw -d "PathOrToken=/owner/bday-of-addressbook/" http://localhost:5232/.sharing/v1/bday/enable
@@ -678,6 +678,34 @@ curl -u owner:$ownerpw -d "PathOrToken=/owner/bday-of-addressbook/" http://local
curl -u owner:$ownerpw -d "PathOrToken=/owner/bday-of-addressbook/" http://localhost:5232/.sharing/v1/bday/unhide
```
* Check
* Fetch *map*
Use e.g. WebUI, an additional (virtual) calendar collection appears
```bash
## Fetch VCALENDAR auto-created from VADDRESSBOOK
curl -u owner:$ownerpw http://localhost:5232/owner/bday-of-addressbook/
BEGIN:VCALENDAR
...
END:VCALENDAR
```
Via WebUI an additional (virtual) calendar collection appears
* Create as *token*
```bash
## Create sharing of type *token* with conversion *bday* (shown PathOrToken is an example)
curl -u owner:$ownerpw -d "Enabled=true" -d "Hidden=false" -d "PathMapped=/owner/addressbook/" -d "User=owner" -d "Conversion=bday" http://localhost:5232/.sharing/v1/token/create
ApiVersion=1
Status='success'
PathOrToken='/.token/v1/lqqwqhZYTGi9uSPsixien_8G5jiSK0FfhNFRGG_t8UA0/'
```
* Fetch *map*
```bash
## Fetch VCALENDAR auto-created from VADDRESSBOOK
curl http://localhost:5232/.token/v1/lqqwqhZYTGi9uSPsixien_8G5jiSK0FfhNFRGG_t8UA0/
BEGIN:VCALENDAR
...
END:VCALENDAR
```

18
contrib/apache/radicale.conf
View File

@@ -50,8 +50,8 @@
RewriteCond %{REQUEST_METHOD} GET
RewriteRule ^/radicale/$ /radicale/.web/ [R,L]
<LocationMatch "^/radicale/\.web.*>
# Internal WebUI does not need authentication at all
<LocationMatch "^/radicale/\.(web|token).*>
# Internal WebUI or Token does not need authentication at all
RequestHeader set X-Script-Name /radicale
RequestHeader set X-Forwarded-Port "%{SERVER_PORT}s"
@@ -69,7 +69,7 @@
</IfDefine>
</LocationMatch>
<LocationMatch "^/radicale(?!/\.web)">
<LocationMatch "^/radicale(?!/\.(web|token))">
RequestHeader set X-Script-Name /radicale
RequestHeader set X-Forwarded-Port "%{SERVER_PORT}s"
@@ -133,8 +133,8 @@
WSGIScriptAlias /radicale /usr/share/radicale/radicale.wsgi
# Internal WebUI does not need authentication at all
<LocationMatch "^/radicale/\.web.*>
# Internal WebUI or Token does not need authentication at all
<LocationMatch "^/radicale/\.(web|token).*>
RequestHeader set X-Script-Name /radicale
Require local
@@ -143,7 +143,7 @@
</IfDefine>
</LocationMatch>
<LocationMatch "^/radicale(?!/\.web)">
<LocationMatch "^/radicale(?!/\.(web|token))">
RequestHeader set X-Script-Name /radicale
<IfDefine !RADICALE_SERVER_USER_AUTHENTICATION>
@@ -221,7 +221,7 @@ CustomLog logs/ssl_request_log "%t %h %{SSL_PROTOCOL}x %{SSL_CIPHER}x \"%r\" %b"
RewriteCond %{REQUEST_METHOD} GET
RewriteRule ^/$ /.web/ [R,L]
<LocationMatch "^/\.web.*>
<LocationMatch "^/\.(web|token).*>
RequestHeader set X-Forwarded-Port "%{SERVER_PORT}s"
RequestHeader set X-Forwarded-Proto expr=%{REQUEST_SCHEME}
@@ -237,7 +237,7 @@ CustomLog logs/ssl_request_log "%t %h %{SSL_PROTOCOL}x %{SSL_CIPHER}x \"%r\" %b"
</IfDefine>
</LocationMatch>
<LocationMatch "^(?!/\.web)">
<LocationMatch "^(?!/\.(web|token))">
RequestHeader set X-Forwarded-Port "%{SERVER_PORT}s"
RequestHeader set X-Forwarded-Proto expr=%{REQUEST_SCHEME}
@@ -292,7 +292,7 @@ CustomLog logs/ssl_request_log "%t %h %{SSL_PROTOCOL}x %{SSL_CIPHER}x \"%r\" %b"
WSGIScriptAlias / /usr/share/radicale/radicale.wsgi
<LocationMatch "^/(?!/\.web)">
<LocationMatch "^/(?!/\.(web|token))">
<IfDefine !RADICALE_SERVER_USER_AUTHENTICATION>
## User authentication handled by "radicale"
Require local

8
contrib/caddy/radicale.caddyfile
View File

@@ -11,13 +11,13 @@ caldav.example.com {
redir @get-root /.web/
# Do not auth on /.web/*
@not-webui {
not path /.web/*
# Do not auth on /.web/* or /.token/*
@not-auth {
not path /.web/* /.token/*
}
# disable this in case authentication is handled by Radicale
basic_auth @not-webui {
basic_auth @not-auth {
USER HASH
}

6
radicale/app/get.py
View File

@@ -31,7 +31,7 @@ from radicale.log import logger
def propose_filename(collection: storage.BaseCollection, share: Union[dict, None] = None) -> str:
"""Propose a filename for a collection."""
share_bday_automap = False
if share and share['ShareType'] == "bday":
if share and share['Conversion'] == "bday":
share_bday_automap = True
if collection.tag == "VADDRESSBOOK" and not share_bday_automap:
fallback_title = "Address book"
@@ -112,7 +112,7 @@ class ApplicationPartGet(ApplicationBase):
if not item.tag:
return (httputils.NOT_ALLOWED if limited_access else
httputils.DIRECTORY_LISTING)
if share and share['ShareType'] == "bday":
if share and share['Conversion'] == "bday":
content_type = xmlutils.MIMETYPES["VCALENDAR"]
else:
content_type = xmlutils.MIMETYPES[item.tag]
@@ -130,7 +130,7 @@ class ApplicationPartGet(ApplicationBase):
"ETag": item.etag}
if content_disposition:
headers["Content-Disposition"] = content_disposition
if isinstance(item, storage.BaseCollection) and self._sharing._enabled and share and share['ShareType'] == "bday":
if isinstance(item, storage.BaseCollection) and share and share['Conversion'] == "bday":
# convert VCF to ICS
answer = item.serialize(vcf_to_ics=True)
else:

40
radicale/app/propfind.py
View File

@@ -73,11 +73,11 @@ def xml_propfind(base_prefix: str, path: str,
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND/xml_propfind: shares=%r", shares)
for item, permission, sharetype in allowed_items:
for item, permission, conversion in allowed_items:
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, shares=shares, sharetype=sharetype))
allprop=allprop, propname=propname, max_resource_size=max_resource_size, shares=shares, conversion=conversion))
return multistatus
@@ -85,7 +85,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, shares: dict = {}, sharetype: Union[str, None] = None) -> ET.Element:
propname: bool = False, allprop: bool = False, shares: dict = {}, conversion: Union[str, 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")
@@ -109,7 +109,7 @@ def xml_propfind_response(
# lookup share
share = None
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND/xml_propfind: sharetype=%r item.path=%r", sharetype, uri)
logger.debug("TRACE/PROPFIND/xml_propfind: conversion=%r item.path=%r", conversion, uri)
for entry in shares:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND/xml_propfind: check entry=%r", entry)
@@ -117,19 +117,20 @@ def xml_propfind_response(
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND/xml_propfind: PathMapped=%r uri=%r", shares[entry]['PathMapped'], uri)
if uri.startswith(shares[entry]['PathMapped']):
if sharetype is not None and shares[entry]['ShareType'] == sharetype:
if conversion is not None and shares[entry]['Conversion'] == conversion:
share = shares[entry]
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND/xml_propfind: found share=%r", share)
break
share_bday_automap = False
if share and share['ShareType'] == "bday":
if share and share['Conversion'] == "bday":
share_bday_automap = True
if share:
# backmap
uri = uri.replace(share['PathMapped'], share['PathOrToken'])
if uri.startswith(share['PathMapped']):
uri = str(share['PathOrToken']) + uri.removeprefix(share['PathMapped'])
if share_bday_automap and not uri.endswith("/"):
uri = uri.rstrip(".vcf") + ".ics"
@@ -217,9 +218,10 @@ def xml_propfind_response(
child_element.text = xmlutils.make_href(base_prefix, path)
if share:
# backmap
child_element.text = child_element.text.replace(share['PathMapped'], share['PathOrToken'])
if share_bday_automap:
child_element.text = child_element.text.rstrip(".vcf") + ".ics"
if child_element.text.startswith(share['PathMapped']):
child_element.text = str(share['PathOrToken']) + child_element.text.removeprefix(share['PathMapped'])
if share_bday_automap and child_element.text.endswith(".vcf"):
child_element.text = child_element.text.removesuffix(".vcf") + ".ics"
element.append(child_element)
elif tag == xmlutils.make_clark("C:supported-calendar-component-set"):
human_tag = xmlutils.make_human_tag(tag)
@@ -256,13 +258,15 @@ def xml_propfind_response(
elif tag == xmlutils.make_clark("D:current-user-principal"):
if user:
child_element = ET.Element(xmlutils.make_clark("D:href"))
child_element.text = xmlutils.make_href(
base_prefix, "/%s/" % user)
if share:
# backmap
child_element.text = child_element.text.replace(share['Owner'], share['User'])
if share_bday_automap:
child_element.text = child_element.text.rstrip(".vcf") + ".ics"
child_element.text = xmlutils.make_href(
base_prefix, "/%s/" % share['User'])
if share_bday_automap and child_element.text.endswith(".vcf"):
child_element.text = child_element.text.removesuffix(".vcf") + ".ics"
else:
child_element.text = xmlutils.make_href(
base_prefix, "/%s/" % user)
element.append(child_element)
else:
element.append(ET.Element(
@@ -546,10 +550,10 @@ class ApplicationPartPropfind(ApplicationBase):
items_iter = itertools.chain([item], items_iter)
for item, permission in list(self._collect_allowed_items(items_iter, user)):
if self._sharing._enabled and share:
if share['ShareType'] == "bday" and not isinstance(item, storage.BaseCollection):
if share['Conversion'] == "bday" and not isinstance(item, storage.BaseCollection):
if not item.convert_vcf_to_ics():
continue
allowed_items.append((item, permission, share['ShareType']))
allowed_items.append((item, permission, share['Conversion']))
else:
allowed_items.append((item, permission, None))
if self._sharing._enabled:
@@ -577,7 +581,7 @@ class ApplicationPartPropfind(ApplicationBase):
c_items_iter = iter(self._storage.discover(c_path, "0"))
c_allowed_items = list(self._collect_allowed_items(c_items_iter, c_user))
for item, permission in c_allowed_items:
allowed_items.append((item, permission, share['ShareType']))
allowed_items.append((item, permission, share['Conversion']))
shares[c_share] = share
headers = {"DAV": httputils.DAV_HEADERS,

28
radicale/app/report.py
View File

@@ -161,7 +161,7 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
logger.debug("TRACE/REPORT/xml_report: base_prefix=%r path=%r", base_prefix, path)
share_bday_automap = False
if share and share['ShareType'] == "bday":
if share and share['Conversion'] == "bday":
share_bday_automap = True
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report(1): share=%r", share)
@@ -263,7 +263,7 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
collection_tag = collection.tag
# !!! Don't access storage after this !!!
unlock_storage_fn()
if share and share['ShareType'] == "bday":
if share and share['Conversion'] == "bday":
collection_tag = "VCALENDAR"
# autoconvert
retrieved_items_vcf_to_ics = []
@@ -740,7 +740,7 @@ def xml_item_response(base_prefix: str, href: str,
logger.debug("TRACE/REPORT/xml_item_response: found=%s share=%r", found_item, share)
share_bday_automap = False
if share and share['ShareType'] == "bday":
if share and share['Conversion'] == "bday":
share_bday_automap = True
href_element = ET.Element(xmlutils.make_clark("D:href"))
@@ -749,9 +749,12 @@ def xml_item_response(base_prefix: str, href: str,
logger.debug("TRACE/REPORT/xml_report: href=%r", href_element.text)
if share:
# backmap
href_element.text = href_element.text.replace(share['PathMapped'], share['PathOrToken'])
if share_bday_automap:
href_element.text = href_element.text.rstrip(".vcf") + ".ics"
if href_element.text.startswith(share['PathMapped']):
href_element.text = str(share['PathOrToken']) + href_element.text.removeprefix(share['PathMapped'])
if share_bday_automap and href_element.text.endswith(".vcf"):
href_element.text = href_element.text.removesuffix(".vcf") + ".ics"
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: href=%r (backmapped)", href_element.text)
response.append(href_element)
if found_item:
@@ -790,12 +793,17 @@ def retrieve_items(
gets set to ``True``."""
nonlocal collection_requested
for hreference in hreferences:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: hreference=%r", hreference)
if share:
# map back to owner
hreference = hreference.replace(share['PathOrToken'], share['PathMapped'])
if share['ShareType'] == "bday":
if not hreference.endswith('/'):
hreference = hreference.rstrip(".ics") + ".vcf"
if hreference.startswith(share['PathOrToken']):
hreference = str(share['PathMapped']) + hreference.removeprefix(share['PathOrToken'])
if share['Conversion'] == "bday":
if hreference.endswith(".ics"):
hreference = hreference.removesuffix(".ics") + ".vcf"
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/REPORT/xml_report: hreference=%r (backmapped)", hreference)
try:
name = pathutils.name_from_path(hreference, collection)
except ValueError as e:

16
radicale/config.py
View File

@@ -487,10 +487,10 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
"value": "false",
"help": "enable sharing of collection by map",
"type": bool}),
("collection_by_bday", {
"value": "false",
"help": "enable sharing of collection by bday (conversion on-the-fly)",
"type": bool}),
("collection_by_bday", { # TODO: remove/3.7.0-final
"value": "false", # TODO: remove/3.7.0-final
"help": "enable sharing of collection by bday (conversion on-the-fly)", # TODO: remove/3.7.0-final
"type": bool}), # TODO: remove/3.7.0-final
("permit_create_token", {
"value": "false",
"help": "permit create of token-based sharing",
@@ -499,10 +499,10 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
"value": "false",
"help": "permit create of map-based sharing",
"type": bool}),
("permit_create_bday", {
"value": "false",
"help": "permit create of bday-based sharing",
"type": bool}),
("permit_create_bday", { # TODO: remove/3.7.0-final
"value": "false", # TODO: remove/3.7.0-final
"help": "permit create of bday-based sharing", # TODO: remove/3.7.0-final
"type": bool}), # TODO: remove/3.7.0-final
("permit_properties_overlay", {
"value": "false",
"help": "permit properties overlay",

4
radicale/rights/__init__.py
View File

@@ -35,8 +35,6 @@ Permissions:
- t: deny create of token-based sharing of collection in case permit_create_token=True (>= 3.7.0)
- M: permit create of map-based sharing of collection in case permit_create_map=False (>= 3.7.0)
- m: deny create of map-based sharing of collection in case permit_create_map=True (>= 3.7.0)
- B: permit create of bday-based sharing of addressbook collection in case permit_create_bday=False (>= 3.7.0)
- b: deny create of bday-based sharing of addressbook collection in case permit_create_bday=True (>= 3.7.0)
Permissions only supported so far in share permissions:
- P: permit properties overlay in case permit_properties_overlay=False (>= 3.7.0)
@@ -55,7 +53,7 @@ from radicale import config, utils
INTERNAL_TYPES: Sequence[str] = ("authenticated", "owner_write", "owner_only",
"from_file")
INTERNAL_PERMISSIONS: str = "RriWwDdOoTtMmPpEeBb"
INTERNAL_PERMISSIONS: str = "RriWwDdOoTtMmPpEe"
def load(configuration: "config.Configuration") -> "BaseRights":

352
radicale/sharing/__init__.py
View File

@@ -16,9 +16,11 @@
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
import base64
import copy # TODO: remove/3.7.0-final
import io
import json
import logging
import os # TODO: remove/3.7.0-final
import re
import socket
import uuid
@@ -34,7 +36,7 @@ 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', 'Properties')
DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Conversion', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated', 'Properties', 'Actions')
# ShareType: <token|map>
# PathOrToken: <path|token> [PrimaryKey]
# PathMapped: <path>
@@ -48,6 +50,10 @@ DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner'
# TimestampCreated: <unixtime> (when created)
# TimestampUpdated: <unixtime> (last update)
# Properties: Overlay of collection properties in JSON
# Conversion: none|bday
# bday: check VADDRESSBOOK VCARD(vcf) entries for BDAY and convert to VCALENDAR reoccuring VEVENT(ics)
# Actions: Actions structure in JSON
# (future reserved for e.g. "filter", "filter_pre", "filter_post" or anything else, implemented on request)
DB_TYPES_V1: dict[str, type] = {
"ShareType": str,
@@ -62,19 +68,24 @@ DB_TYPES_V1: dict[str, type] = {
"HiddenByUser": bool,
"TimestampCreated": int,
"TimestampUpdated": int,
"Properties": dict
"Properties": dict,
"Conversion": str,
"Actions": dict,
}
DB_FIELDS_V1_USER_PERMITTED: Sequence[str] = ('EnabledByUser', 'HiddenByUser', 'Properties')
SHARE_TYPES: Sequence[str] = ('token', 'map', 'bday', 'all')
SHARE_TYPES_V1: Sequence[str] = ('token', 'map', 'bday')
SHARE_TYPES: Sequence[str] = ('token', 'map', 'all')
# token: share by secret token (does not require authentication)
# map : share by mapping collection of one user to another as virtual
# bday : share by mapping addressbook-collection of one user to another as virtual calendar-collection
# all : only supported for "list" and "info"
SHARE_TYPES_V1: Sequence[str] = ('token', 'map')
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
SHARE_TYPES: Sequence[str] = ('token', 'map', 'bday', 'all') # type: ignore[no-redef] # TODO: remove/3.7.0-final
SHARE_TYPES_V1: Sequence[str] = ('token', 'map', 'bday') # type: ignore[no-redef] # TODO: remove/3.7.0-final
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
@@ -106,9 +117,14 @@ API_TYPES_V1: dict[str, type] = {
"Permissions": str,
"Enabled": bool,
"Hidden": bool,
"Properties": dict}
"Properties": dict,
"Conversion": str,
"Actions": dict,
"SupportedConversions": list,
}
TOKEN_PATTERN_V1: str = "(v1/[a-zA-Z0-9_=\\-]{44})"
TOKEN_PATTERN_V1: str = "v1/[a-zA-Z0-9_\\-]{44}"
PATH_PATTERN: str = "([a-zA-Z0-9/.\\-]+)" # TODO: extend or find better source
@@ -116,6 +132,8 @@ USER_PATTERN: str = "([a-zA-Z0-9@]+)" # TODO: extend or find better source
OVERLAY_PROPERTIES_WHITELIST: Sequence[str] = ("C:calendar-description", "ICAL:calendar-color", "CR:addressbook-description", "INF:addressbook-color", "D:displayname")
CONVERSIONS_WHITELIST: Sequence[str] = ("bday", "none")
def load(configuration: "config.Configuration") -> "BaseSharing":
"""Load the sharing database module chosen in configuration."""
@@ -145,20 +163,21 @@ class BaseSharing:
# Sharing
self.sharing_collection_by_map = configuration.get("sharing", "collection_by_map")
self.sharing_collection_by_token = configuration.get("sharing", "collection_by_token")
self.sharing_collection_by_bday = configuration.get("sharing", "collection_by_bday")
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
self.sharing_collection_by_map = self.sharing_collection_by_map or configuration.get("sharing", "collection_by_bday") # TODO: remove/3.7.0-final
self.permit_create_token = configuration.get("sharing", "permit_create_token")
self.permit_create_map = configuration.get("sharing", "permit_create_map")
self.permit_create_bday = configuration.get("sharing", "permit_create_bday")
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
self.permit_create_map = self.permit_create_map or configuration.get("sharing", "permit_create_bday") # TODO: remove/3.7.0-final
self.default_permissions_create_token = configuration.get("sharing", "default_permissions_create_token")
self.default_permissions_create_map = configuration.get("sharing", "default_permissions_create_map")
self.permit_properties_overlay = configuration.get("sharing", "permit_properties_overlay")
self.enforce_properties_overlay = configuration.get("sharing", "enforce_properties_overlay")
logger.info("sharing.collection_by_map : %s", self.sharing_collection_by_map)
logger.info("sharing.collection_by_token: %s", self.sharing_collection_by_token)
logger.info("sharing.collection_by_bday : %s", self.sharing_collection_by_bday)
logger.info("sharing.permit_create_token: %s", self.permit_create_token)
logger.info("sharing.permit_create_map : %s", self.permit_create_map)
logger.info("sharing.permit_create_bday : %s", self.permit_create_bday)
logger.info("sharing.default_permissions_create_token: %r", self.default_permissions_create_token)
logger.info("sharing.default_permissions_create_map : %r", self.default_permissions_create_map)
logger.info("sharing.permit_properties_overlay: %s", self.permit_properties_overlay)
@@ -168,7 +187,7 @@ class BaseSharing:
self.sharing_db_type = configuration.get("sharing", "type")
logger.info("sharing.database_type: %s", self.sharing_db_type)
if ((self.sharing_collection_by_map is False) and (self.sharing_collection_by_token is False) and (self.sharing_collection_by_bday is False)):
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
@@ -183,7 +202,7 @@ class BaseSharing:
"""
try:
if self.database_init() is False:
logger.info("sharing disabled as no database is active")
logger.warning("sharing disabled as no database is active")
self._enabled = False
return False
except Exception as e:
@@ -216,9 +235,11 @@ class BaseSharing:
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]:
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Conversion: Union[str, None] = None,
) -> list[dict]:
""" retrieve sharing """
return []
@@ -233,12 +254,15 @@ class BaseSharing:
def database_create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Conversion: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
Properties: Union[dict, None] = None,
Actions: Union[dict, None] = None,
) -> dict:
""" create sharing """
return {"status": "not-implemented"}
@@ -254,7 +278,10 @@ class BaseSharing:
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
Properties: Union[dict, None] = None,
Conversion: Union[str, None] = None,
Actions: Union[dict, None] = None,
) -> dict:
""" update sharing """
return {"status": "not-implemented"}
@@ -330,7 +357,12 @@ class BaseSharing:
# *** sharing functions called by request methods ***
# list sharings
def sharing_collection_list(self, User: Union[str, None] = None, Enabled: Union[bool, None] = None, Hidden: Union[bool, None] = None) -> list[dict]:
def sharing_collection_list(self,
User: Union[str, None] = None,
Enabled: Union[bool, None] = None,
Hidden: Union[bool, None] = None,
Conversion: Union[str, None] = None,
) -> list[dict]:
""" returning dict with shared collections by filter(User/Enabled/Hidden) or None if not found"""
sharing_collection_list = []
@@ -346,27 +378,16 @@ class BaseSharing:
EnabledByOwner=Enabled,
EnabledByUser=Enabled,
HiddenByOwner=Hidden,
HiddenByUser=Hidden)
if not self.sharing_collection_by_bday:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday: not active")
else:
# retrieve collections depending on filter
sharing_collection_list += self.database_list_sharing(
ShareType="bday",
OwnerOrUser=User,
User=User,
EnabledByOwner=Enabled,
EnabledByUser=Enabled,
HiddenByOwner=Hidden,
HiddenByUser=Hidden)
HiddenByUser=Hidden,
Conversion=Conversion,
)
return sharing_collection_list
# resolves a path to a share
def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
logger.debug("TRACE/sharing/resolver: lookup path=%r user=%r", path, user)
share = None
if path == "/":
@@ -376,6 +397,8 @@ class BaseSharing:
if self.sharing_collection_by_token:
if share is None:
share = self.sharing_collection_by_token_resolver(path)
if share is not None and 'error' in share:
return None
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: not active")
@@ -383,17 +406,12 @@ class BaseSharing:
if self.sharing_collection_by_map:
if share is None:
share = self.sharing_collection_by_map_resolver(path, user)
if share is not None and 'error' in share:
return None
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
if self.sharing_collection_by_bday:
if share is None:
share = self.sharing_collection_by_bday_resolver(path, user)
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday: not active")
if share is not None:
if self.permit_properties_overlay:
if share['Permissions'] and "p" not in share['Permissions']:
@@ -426,31 +444,44 @@ class BaseSharing:
# *** internal sharing functions ***
# resolves a token "path" to a share
# dict: share
# None: not supported
# False: supported but not found
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)
logger.debug("TRACE/sharing/token/resolver: check path: %r", path)
if path.startswith("/.token/"):
pattern = re.compile('^/\\.token/' + TOKEN_PATTERN_V1 + '$')
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
logger.debug("TRACE/sharing/token/resolver: unsupported token: %r", path)
return {'error': 'token-not-supported'}
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])
logger.debug("TRACE/sharing/token/resolver: supported token: %r", path)
result = self.database_get_sharing(
ShareType="token",
OnlyEnabled=False,
PathOrToken=match[1])
if result is not None:
logger.info("Sharing/%s: resolved %r->%r, user ->%r, permissions %r", "token", path, result['PathMapped'], result['Owner'], result['Permissions'])
if result is None:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token/resolver: supported token not found: %r", path)
return {'error': 'token-not-found'}
if result['EnabledByOwner'] is not True:
logger.info("Sharing/%s: resolved path %r->%r, User=%r not enabled by owner", "token", path, result['PathMapped'], result['Owner'])
return {'error': 'token-not-enabled'}
logger.info("Sharing/%s: resolved %r->%r, User=%r, Permissions=%r Conversion=%r", "token", path, result['PathMapped'], result['Owner'], result['Permissions'], result['Conversion'])
return result
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: no supported prefix found in path: %r", path)
logger.debug("TRACE/sharing/token/resolver: no supported prefix found in path: %r", path)
return None
else:
if logger.isEnabledFor(logging.DEBUG):
@@ -463,13 +494,14 @@ class BaseSharing:
if self.sharing_collection_by_map:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/resolver: check path: %r", path)
result = self.database_get_sharing(
ShareType="map",
PathOrToken=path,
User=user)
if result:
pass
else:
ShareType="map",
PathOrToken=path,
OnlyEnabled=False,
User=user)
if not result:
# fallback to parent path
parent_path = pathutils.parent_path(path)
if logger.isEnabledFor(logging.DEBUG):
@@ -477,6 +509,7 @@ class BaseSharing:
result = self.database_get_sharing(
ShareType="map",
PathOrToken=parent_path,
OnlyEnabled=False,
User=user)
if result:
result['PathMapped'] = path.replace(parent_path, result['PathMapped'])
@@ -484,53 +517,26 @@ class BaseSharing:
logger.debug("TRACE/sharing/map/resolver: PathMapped=%r Permissions=%r by parent_path=%r", result['PathMapped'], result['Permissions'], parent_path)
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not found")
logger.debug("TRACE/sharing/map/resolver: not found")
return None
logger.info("Sharing/%s: resolved path %r->%r, user %r->%r, permissions %r", "map", path, result['PathMapped'], user, result['Owner'], result['Permissions'])
return result
if result:
if result['EnabledByOwner'] is not True:
logger.info("Sharing/%s: resolved path %r->%r, user %r->%r not enabled by owner", "map", path, result['PathMapped'], user, result['Owner'])
return {'error': 'map-not-enabled'}
if result['EnabledByUser'] is not True:
logger.info("Sharing/%s: resolved path %r->%r, user %r->%r not enabled by user", "map", path, result['PathMapped'], user, result['Owner'])
return {'error': 'map-not-enabled'}
logger.info("Sharing/%s: resolved path %r->%r, user %r->%r, Permissions=%r Conversion=%r", "map", path, result['PathMapped'], user, result['Owner'], result['Permissions'], result['Conversion'])
return result
return None
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
return None
# resolves a bday "path" to a share
def sharing_collection_by_bday_resolver(self, path: str, user: str) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
if self.sharing_collection_by_bday:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday/resolver: check path: %r", path)
result = self.database_get_sharing(
ShareType="bday",
PathOrToken=path,
User=user)
if result:
pass
else:
# fallback to parent path
parent_path = pathutils.parent_path(path)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday/resolver: check parent path: %r", parent_path)
result = self.database_get_sharing(
ShareType="bday",
PathOrToken=parent_path,
User=user)
if result:
result['PathMapped'] = path.replace(parent_path, result['PathMapped'])
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday/resolver: PathMapped=%r Permissions=%r by parent_path=%r", result['PathMapped'], result['Permissions'], parent_path)
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday: not found")
return None
logger.info("Sharing/%s: resolved path %r->%r, user %r->%r, permissions %r", "bday", path, result['PathMapped'], user, result['Owner'], result['Permissions'])
return result
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday: 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
@@ -545,27 +551,28 @@ class BaseSharing:
``user`` is empty for anonymous users.
Request:
action: (token|map|bday)/list
action: (token|map)/list
PathOrToken: <path|token> (optional for filter)
action: (token|map|bday)/create
action: (token|map)/create
PathMapped: <path> (mandatory)
Permissions: <Permissions> (default: r)
token -> returns <token>
map|bday
map
PathOrToken: <path> (mandatory)
User: <target_user> (mandatory)
Conversion: None|bday (optional)
action: (token|map|bday)/update
action: (token|map)/update
action: (token|map|bday)/(delete|disable|enable|hide|unhide)
action: (token|map)/(delete|disable|enable|hide|unhide)
PathOrToken: <path|token> (mandatory)
token
map|bday
map
PathMapped: <path> (mandatory)
User: <target_user>
@@ -626,11 +633,6 @@ class BaseSharing:
logger.warning(api_info + ": not enabled by config (collection_by_map)")
return httputils.NOT_FOUND
if not self.sharing_collection_by_bday and ShareType == "bday":
# API "token" is not enabled
logger.warning(api_info + ": not enabled by config (collection_by_bday)")
return httputils.NOT_FOUND
# check for valid API hooks
if action not in API_HOOKS_V1:
if logger.isEnabledFor(logging.DEBUG):
@@ -738,12 +740,14 @@ class BaseSharing:
# parameters default
PathOrToken: Union[str, None] = None
PathMapped: Union[str, None] = None
User: Union[str, None] = None
PathMapped: Union[str, None] = None
User: Union[str, None] = None
Permissions: Union[str, None] = None # no permissions by default
Enabled: Union[bool, None] = None
Hidden: Union[bool, None] = None
Properties: Union[dict, None] = None
Enabled: Union[bool, None] = None
Hidden: Union[bool, None] = None
Properties: Union[dict, None] = None
Conversion: Union[str, None] = None
Actions: Union[dict, None] = None # reserved so far
# parameters sanity check
for key in request_data:
@@ -753,7 +757,7 @@ class BaseSharing:
return httputils.bad_request("Invalid value for Permissions")
elif key == "PathOrToken":
if ShareType == "token":
if not re.search('^' + TOKEN_PATTERN_V1 + '$', request_data[key]):
if not re.search('^/.token/' + TOKEN_PATTERN_V1 + '/$', request_data[key]):
logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken")
else:
@@ -776,7 +780,11 @@ class BaseSharing:
# check for optional parameters
if 'PathMapped' in request_data:
# used by create or list(filter)
PathMapped = request_data['PathMapped']
if base_prefix:
PathMapped = request_data['PathMapped'].removeprefix(base_prefix)
logger.debug(api_info + ": remove base_prefix PathMapped=%r->%r", request_data['PathMapped'], PathMapped)
else:
PathMapped = request_data['PathMapped']
if 'PathOrToken' not in request_data:
if action == 'info':
@@ -805,6 +813,15 @@ class BaseSharing:
return httputils.bad_request("Property not supported to overlay: %r" % entry)
Properties = request_data['Properties']
if 'Conversion' in request_data:
Conversion = request_data['Conversion']
# verify against whitelist
if Conversion not in CONVERSIONS_WHITELIST:
return httputils.bad_request("Conversion not supported: %r" % Conversion)
if 'Actions' in request_data:
return httputils.bad_request("Actions currently not supported (reserved for future needs)")
if 'Enabled' in request_data:
Enabled = request_data['Enabled']
else:
@@ -826,7 +843,12 @@ class BaseSharing:
answer['ApiVersion'] = 1
Timestamp = int((datetime.now() - datetime(1970, 1, 1)).total_seconds())
if not self.sharing_collection_by_map and not self.sharing_collection_by_token and not self.sharing_collection_by_bday:
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
if ShareType == "bday": # TODO: remove/3.7.0-final
ShareType = "map" # TODO: remove/3.7.0-final
Conversion = "bday" # TODO: remove/3.7.0-final
if not self.sharing_collection_by_map and not self.sharing_collection_by_token:
if not action == 'info':
# API is not enabled
logger.warning(api_info + ": API is not enabled")
@@ -846,12 +868,24 @@ class BaseSharing:
ShareType=ShareType,
OwnerOrUser=user,
PathMapped=PathMapped,
PathOrToken=PathOrToken)
PathOrToken=PathOrToken,
Conversion=Conversion,
)
else:
result_array = self.database_list_sharing(
OwnerOrUser=user,
PathMapped=PathMapped,
PathOrToken=PathOrToken)
PathOrToken=PathOrToken,
Conversion=Conversion,
)
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
# check and change to legacy ShareType # TODO: remove/3.7.0-final
result_array_adj = copy.deepcopy(result_array) # TODO: remove/3.7.0-final
for index in range(0, len(result_array_adj)): # TODO: remove/3.7.0-final
if result_array_adj[index]['Conversion'] == "bday": # TODO: remove/3.7.0-final
result_array_adj[index]['ShareType'] = "bday" # TODO: remove/3.7.0-final
result_array = result_array_adj
answer['Lines'] = len(result_array)
if len(result_array) == 0:
@@ -871,6 +905,9 @@ class BaseSharing:
logger.warning(api_info + ": missing PathMapped")
return httputils.bad_request("Missing PathMapped")
if Conversion is None:
Conversion = "none"
# check whether collection exists
with self._storage.acquire_lock("r", user, path=PathMapped):
item = next(iter(self._storage.discover(PathMapped)), None)
@@ -878,7 +915,12 @@ class BaseSharing:
logger.warning(api_info + ": cannot find PathMapped=%r", PathMapped)
return httputils.NOT_FOUND
if not isinstance(item, storage.BaseCollection):
logger.warning(api_info + ": PathMapped=%r is not a collection", PathMapped)
return httputils.METHOD_NOT_ALLOWED
if Conversion == "bday":
if item.tag != "VADDRESSBOOK":
logger.warning(api_info + ": PathMapped=%r is not a VADDRESSBOOK collection (mandatory for Conversion=%r)", PathMapped, Conversion)
return httputils.METHOD_NOT_ALLOWED
if Permissions is None:
if ShareType == "token":
@@ -890,6 +932,12 @@ class BaseSharing:
Permissions = "r"
else:
Permissions = str(Permissions)
if Conversion == "bday":
# bday is read-only
for permission in Permissions:
if permission not in "r":
logger.warning(api_info + ": PathMapped=%r Permissions=%r not supported for Conversion=%r", PathMapped, Permissions, Conversion)
return httputils.METHOD_NOT_ALLOWED
if Enabled is None:
Enabled = False # security by default
@@ -928,8 +976,8 @@ class BaseSharing:
else:
User = user
# v1: create uuid token with 2x 32 bytes = 256 bit
token = "v1/" + str(base64.urlsafe_b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes), 'utf-8')
# v1: create uuid token with 2x 32 bytes = 256 bit with base64 encoding but replace '=' with '0' to avoid any additional encoding issues issues
token = "/.token/v1/" + str(base64.urlsafe_b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes), 'utf-8').replace('=', '0') + "/"
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": %r (Permissions=%r token=%r)", PathMapped, Permissions, token)
@@ -946,12 +994,15 @@ class BaseSharing:
HiddenByOwner=Hidden,
HiddenByUser=HiddenByUser,
Timestamp=Timestamp,
Properties=Properties)
Properties=Properties,
Conversion=Conversion,
Actions=Actions,
)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": result=%r", result)
elif ShareType in ["map", "bday"]:
elif ShareType == "map":
# check preconditions
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
@@ -969,10 +1020,10 @@ class BaseSharing:
else:
User = str(User)
# lookup existing shares with requested PathMapped for same User
shares = self.database_list_sharing(ShareType=ShareType, PathMapped=PathMapped, User=User)
# lookup existing shares with requested PathMapped for same User and same Conversion
shares = self.database_list_sharing(ShareType=ShareType, PathMapped=PathMapped, User=User, Conversion=Conversion)
if len(shares) > 0:
logger.warning(api_info + ": share already exists with PathMapped=%r User=%r", PathMapped, User)
logger.warning(api_info + ": share already exists with PathMapped=%r User=%r Conversion=%r", PathMapped, User, Conversion)
return httputils.CONFLICT
# check access Permissions
@@ -981,25 +1032,14 @@ class BaseSharing:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r", PathMapped, user)
return httputils.NOT_ALLOWED
if ShareType == "map":
if self.permit_create_map is False:
if "m" not in access.permissions:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=False but explicit grant misses 'm')", PathMapped, user)
return httputils.NOT_ALLOWED
else:
if "M" in access.permissions:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'M')", PathMapped, user)
return httputils.NOT_ALLOWED
elif ShareType == "bday":
if self.permit_create_bday is False:
if "b" not in access.permissions:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=False but explicit grant misses 'b')", PathMapped, user)
return httputils.NOT_ALLOWED
else:
if "B" in access.permissions:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'B')", PathMapped, user)
return httputils.NOT_ALLOWED
if self.permit_create_map is False:
if "m" not in access.permissions:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=False but explicit grant misses 'm')", PathMapped, user)
return httputils.NOT_ALLOWED
else:
if "M" in access.permissions:
logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'M')", PathMapped, user)
return httputils.NOT_ALLOWED
access = Access(self._rights, User, PathOrToken)
if not access.check("r"):
@@ -1030,7 +1070,10 @@ class BaseSharing:
HiddenByOwner=Hidden,
HiddenByUser=HiddenByUser,
Timestamp=Timestamp,
Properties=Properties)
Properties=Properties,
Conversion=Conversion,
Actions=Actions,
)
else:
logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
@@ -1050,7 +1093,10 @@ class BaseSharing:
if ShareType == "token":
PathOrToken = token
answer['PathOrToken'] = token
if base_prefix:
answer['PathOrToken'] = base_prefix + token
else:
answer['PathOrToken'] = token
logger.info(api_info + " success: PathMapped=%r Permissions=%r PathOrToken=%r", PathMapped, Permissions, PathOrToken)
@@ -1059,7 +1105,7 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if ShareType not in ["token", "map", "bday"]:
if ShareType not in SHARE_TYPES_V1:
logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
@@ -1185,7 +1231,7 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if ShareType not in ["token", "map", "bday"]:
if ShareType not in SHARE_TYPES_V1:
logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
@@ -1227,19 +1273,21 @@ class BaseSharing:
if ShareType in ["all", "map"]:
answer['FeatureEnabledCollectionByMap'] = self.sharing_collection_by_map
answer['PermittedCreateCollectionByMap'] = self.permit_create_map
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
answer['FeatureEnabledCollectionByBday'] = self.sharing_collection_by_map # TODO: remove/3.7.0-final
answer['PermittedCreateCollectionByBday'] = self.permit_create_map # TODO: remove/3.7.0-final
if ShareType in ["all", "token"]:
answer['FeatureEnabledCollectionByToken'] = self.sharing_collection_by_token
answer['PermittedCreateCollectionByToken'] = self.permit_create_token
if ShareType in ["all", "bday"]:
answer['FeatureEnabledCollectionByBday'] = self.sharing_collection_by_bday
answer['PermittedCreateCollectionByBday'] = self.permit_create_bday
if ShareType in ["all", "map", "token"]:
answer['SupportedConversions'] = CONVERSIONS_WHITELIST
# action: TOGGLE
elif action in API_SHARE_TOGGLES_V1:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/POST/" + action)
if ShareType not in ["token", "map", "bday"]:
if ShareType not in SHARE_TYPES_V1:
logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
@@ -1324,6 +1372,8 @@ class BaseSharing:
if key != 'Content':
if API_TYPES_V1[key] is bool or API_TYPES_V1[key] is int:
answer_array.append(key + '=' + str(answer[key]))
elif API_TYPES_V1[key] is list:
answer_array.append(key + '=(' + str(" ".join(answer[key])) + ')')
else:
answer_array.append(key + "='" + str(answer[key]) + "'")
if 'Content' in answer and answer['Content'] is not None:

51
radicale/sharing/csv.py
View File

@@ -131,8 +131,14 @@ class Sharing(sharing.BaseSharing):
Permissions = row['Permissions']
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
Properties: Union[dict, None] = None
Conversion: Union[str, None] = None
Actions: Union[dict, None] = None
if 'Properties' in row:
Properties = row['Properties']
if 'Conversion' in row:
Conversion = row['Conversion']
if 'Actions' in row:
Actions = row['Actions']
return {
"mapped": True,
"ShareType": ShareType,
@@ -141,8 +147,13 @@ class Sharing(sharing.BaseSharing):
"Owner": Owner,
"User": UserShare,
"Hidden": Hidden,
"EnabledByOwner": row['EnabledByOwner'],
"EnabledByUser": row['EnabledByUser'],
"Permissions": Permissions,
"Properties": Properties}
"Properties": Properties,
"Conversion": Conversion,
"Actions": Actions,
}
return None
def database_list_sharing(self,
@@ -154,7 +165,9 @@ class Sharing(sharing.BaseSharing):
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
HiddenByUser: Union[bool, None] = None,
Conversion: Union[str, None] = None,
) -> list[dict]:
""" retrieve sharing """
row: dict
index = 0
@@ -162,7 +175,7 @@ class Sharing(sharing.BaseSharing):
with self._storage.acquire_lock("r", path=self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser)
logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s Conversion=%r", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser, Conversion)
for row in self._sharing_cache:
if index == 0:
@@ -197,6 +210,8 @@ class Sharing(sharing.BaseSharing):
pass
elif HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser:
pass
elif Conversion is not None and row['Conversion'] != Conversion:
pass
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: add : %r", row)
@@ -207,12 +222,15 @@ class Sharing(sharing.BaseSharing):
def database_create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Conversion: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
Properties: Union[dict, None] = None,
Actions: Union[dict, None] = None,
) -> dict:
""" create sharing """
row: dict
@@ -241,17 +259,6 @@ class Sharing(sharing.BaseSharing):
# must be unique systemwide
logger.error("sharing/map/create: entry already exists: PathMapped=%r User=%r", PathMapped, User)
return {"status": "conflict"}
elif ShareType == "bday":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/bday/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'] != "bday":
continue
if row['PathMapped'] == PathMapped and row['User'] == User and row['PathOrToken'] == PathOrToken:
# must be unique systemwide
logger.error("sharing/bday/create: entry already exists: PathMapped=%r User=%r", PathMapped, User)
return {"status": "conflict"}
else:
return {"status": "error"}
@@ -267,7 +274,10 @@ class Sharing(sharing.BaseSharing):
"HiddenByUser": HiddenByUser,
"TimestampCreated": Timestamp,
"TimestampUpdated": Timestamp,
"Properties": Properties}
"Properties": Properties,
"Conversion": Conversion,
"Actions": Actions,
}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/*/create: add row: %r", row)
@@ -293,7 +303,10 @@ class Sharing(sharing.BaseSharing):
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
Properties: Union[dict, None] = None,
Conversion: Union[str, None] = None,
Actions: Union[dict, None] = None,
) -> dict:
""" update sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r PathMapped=%r Properties=%r EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s", ShareType, PathOrToken, OwnerOrUser, PathMapped, Properties, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser)
@@ -336,6 +349,10 @@ class Sharing(sharing.BaseSharing):
self._sharing_cache[index]["HiddenByUser"] = HiddenByUser
if Properties is not None:
self._sharing_cache[index]["Properties"] = Properties
if Conversion is not None:
self._sharing_cache[index]["Conversion"] = Conversion
if Actions is not None:
self._sharing_cache[index]["Actions"] = Actions
# update timestamp
self._sharing_cache[index]["TimestampUpdated"] = Timestamp

40
radicale/sharing/files.py
View File

@@ -124,8 +124,14 @@ class Sharing(sharing.BaseSharing):
Permissions = row['Permissions']
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
Properties: Union[dict, None] = None
Conversion: Union[str, None] = None
Actions: Union[dict, None] = None
if 'Properties' in row:
Properties = row['Properties']
if 'Conversion' in row:
Conversion = row['Conversion']
if 'Actions' in row:
Actions = row['Actions']
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s Properties=%r)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden, Properties)
return {
@@ -136,8 +142,13 @@ class Sharing(sharing.BaseSharing):
"Owner": Owner,
"User": UserShare,
"Hidden": Hidden,
"EnabledByOwner": row['EnabledByOwner'],
"EnabledByUser": row['EnabledByUser'],
"Permissions": Permissions,
"Properties": Properties}
"Properties": Properties,
"Conversion": Conversion,
"Actions": Actions,
}
return None
@@ -150,12 +161,14 @@ class Sharing(sharing.BaseSharing):
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
HiddenByUser: Union[bool, None] = None,
Conversion: Union[str, 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 EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser)
logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r EnabledByOwner=%s EnabledByUser=%s HiddenByOwner=%s HiddenByUser=%s Conversion=%r", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, EnabledByOwner, EnabledByUser, HiddenByOwner, HiddenByUser, Conversion)
for _ShareType in sharing.SHARE_TYPES_V1:
if ShareType is not None and _ShareType != ShareType:
@@ -206,6 +219,8 @@ class Sharing(sharing.BaseSharing):
pass
elif HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser:
pass
elif Conversion is not None and row['Conversion'] != Conversion:
pass
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: add: %r", row)
@@ -216,12 +231,15 @@ class Sharing(sharing.BaseSharing):
def database_create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Conversion: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
Properties: Union[dict, None] = None,
Actions: Union[dict, None] = None,
) -> dict:
""" create sharing """
row: dict
@@ -245,7 +263,10 @@ class Sharing(sharing.BaseSharing):
"HiddenByUser": HiddenByUser,
"TimestampCreated": Timestamp,
"TimestampUpdated": Timestamp,
"Properties": Properties}
"Properties": Properties,
"Conversion": Conversion,
"Actions": Actions,
}
version = DB_VERSION
@@ -275,7 +296,10 @@ class Sharing(sharing.BaseSharing):
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
Properties: Union[dict, None] = None,
Conversion: Union[str, None] = None,
Actions: Union[dict, None] = None,
) -> dict:
""" update sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r User=%r Properties=%r", ShareType, PathOrToken, OwnerOrUser, User, Properties)
@@ -316,6 +340,10 @@ class Sharing(sharing.BaseSharing):
row["HiddenByUser"] = HiddenByUser
if Properties is not None:
row["Properties"] = Properties
if Conversion is not None:
row["Conversion"] = Conversion
if Actions is not None:
row["Actions"] = Actions
# update timestamp
row["TimestampUpdated"] = Timestamp

4
radicale/tests/__init__.py
View File

@@ -90,6 +90,7 @@ class BaseTest:
remote_useragent = kwargs.pop("remote_useragent", None)
remote_host = kwargs.pop("remote_host", None)
content_type = kwargs.pop("content_type", None)
x_forwarded_for = kwargs.pop("x_forwarded_for", 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():
@@ -108,6 +109,8 @@ class BaseTest:
environ["REMOTE_ADDR"] = remote_host
if content_type:
environ["CONTENT_TYPE"] = content_type
if x_forwarded_for:
environ["HTTP_X_FORWARDED_FOR"] = x_forwarded_for
if accept:
environ["HTTP_ACCEPT"] = accept
environ["REQUEST_METHOD"] = method.upper()
@@ -152,6 +155,7 @@ class BaseTest:
prop_responses[human_tag] = (status_code, element)
status = response.find(xmlutils.make_clark("D:status"))
if status is not None:
logging.debug("test: prop_responses")
assert not prop_responses
assert status.text.startswith("HTTP/1.1 ")
status_code = int(status.text.split(" ")[1])

344
radicale/tests/test_sharing.py
View File

@@ -49,31 +49,39 @@ class TestSharingApiSanity(BaseTest):
f.write(htpasswd_content)
# Helper functions
def _sharing_api(self, sharing_type: str, action: str, check: int, login: Union[str, None], data: str, content_type: str, accept: Union[str, None]) -> Tuple[int, Dict[str, str], str]:
def _sharing_api(self, sharing_type: str, action: str, check: int, login: Union[str, None], data: str, content_type: str, accept: Union[str, None], prefix: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
path_base = "/.sharing/v1/" + sharing_type + "/"
_, headers, answer = self.request("POST", path_base + action, check=check, login=login, data=data, content_type=content_type, accept=accept)
if prefix is not None:
path_base = prefix + path_base
_, headers, answer = self.request("POST", path_base + action, check=check, login=login, data=data, content_type=content_type, accept=accept, x_forwarded_for="127.0.0.2")
else:
_, headers, answer = self.request("POST", path_base + action, check=check, login=login, data=data, content_type=content_type, accept=accept)
logging.info("received answer:\n%s", "\n".join(answer.splitlines()))
return _, headers, answer
def _sharing_api_form(self, sharing_type: str, action: str, check: int, login: Union[str, None], form_array: Sequence[str], accept: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
def _sharing_api_form(self, sharing_type: str, action: str, check: int, login: Union[str, None], form_array: Sequence[str], accept: Union[str, None] = None, prefix: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
data = "&".join(form_array)
content_type = "application/x-www-form-urlencoded"
if accept is None:
accept = "text/plain"
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept)
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept, prefix=prefix)
return _, headers, answer
def _sharing_api_json(self, sharing_type: str, action: str, check: int, login: Union[str, None], json_dict: dict, accept: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
def _sharing_api_json(self, sharing_type: str, action: str, check: int, login: Union[str, None], json_dict: dict, accept: Union[str, None] = None, prefix: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
data = json.dumps(json_dict)
content_type = "application/json"
if accept is None:
accept = "application/json"
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept)
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept, prefix=prefix)
return _, headers, answer
def _propfind_allprop(self, path: str, login) -> dict:
def _propfind_allprop(self, path: str, login: str = "", prefix: Union[str, None] = None) -> dict:
propfind_allprop = get_file_content("allprop.xml")
_, responses = self.propfind(path=path, data=propfind_allprop, login=login)
if prefix is not None:
path = prefix + path
_, responses = self.propfind(path=path, data=propfind_allprop, login=login, x_forwarded_for="127.0.0.2")
else:
_, responses = self.propfind(path=path, data=propfind_allprop, login=login)
logging.info("response: %r", responses)
response = responses[path]
assert not isinstance(response, int)
@@ -256,7 +264,6 @@ class TestSharingApiSanity(BaseTest):
"htpasswd_encryption": "plain"},
"sharing": {
"collection_by_map": "True",
"collection_by_bday": "True",
"collection_by_token": "True"},
"rights": {"type": "owner_only"}})
@@ -334,7 +341,6 @@ class TestSharingApiSanity(BaseTest):
# path with valid API and hook and all enabled
self.configure({"sharing": {
"collection_by_map": "True",
"collection_by_bday": "True",
"collection_by_token": "True"}
})
for sharetype in sharing.SHARE_TYPES:
@@ -357,10 +363,8 @@ class TestSharingApiSanity(BaseTest):
# When turning on permission to create
self.configure({"sharing": {
"collection_by_map": "True",
"collection_by_bday": "True",
"collection_by_token": "True",
"permit_create_map": "True",
"permit_create_bday": "True",
"permit_create_token": "True"}
})
logging.info("\n*** check API hook: info/all")
@@ -369,10 +373,9 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['FeatureEnabledCollectionByMap'] is True, f'FeatureEnabledCollectionByMap {db_type}'
assert answer_dict['FeatureEnabledCollectionByToken'] is True, f'FeatureEnabledCollectionByToken {db_type}'
assert answer_dict['FeatureEnabledCollectionByBday'] is True, f'FeatureEnabledCollectionByBday {db_type}'
assert answer_dict['PermittedCreateCollectionByMap'] is True, f'PermittedCreateCollectionByMap {db_type}'
assert answer_dict['PermittedCreateCollectionByToken'] is True, f'PermittedCreateCollectionByToken {db_type}'
assert answer_dict['PermittedCreateCollectionByBday'] is True, f'PermittedCreateCollectionByBday {db_type}'
assert answer_dict['SupportedConversions'] == ["bday", "none"]
def test_sharing_api_list_with_auth(self) -> None:
"""POST/list with authentication."""
@@ -382,10 +385,8 @@ class TestSharingApiSanity(BaseTest):
"sharing": {
"type": "csv",
"permit_create_map": True,
"permit_create_bday": True,
"permit_create_token": True,
"collection_by_map": "True",
"collection_by_bday": "True",
"collection_by_token": "True"},
"logging": {"request_header_on_debug": "true",
"request_content_on_debug": "True"},
@@ -488,7 +489,6 @@ class TestSharingApiSanity(BaseTest):
"permit_create_map": True,
"permit_create_token": True,
"collection_by_map": "True",
"collection_by_bday": "True",
"collection_by_token": "True"},
"logging": {"request_header_on_debug": "False",
"response_content_on_debug": "True",
@@ -684,6 +684,69 @@ class TestSharingApiSanity(BaseTest):
self.delete(path_base1, login="owner:ownerpw")
self.delete(path_base2, login="owner:ownerpw")
def test_sharing_api_token_usage_proxy(self) -> None:
"""share-by-token API tests simulating a reverse proxy - real usage."""
script_name = "/radicale"
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_map": True,
"permit_create_token": True,
"collection_by_map": "True",
"collection_by_token": "True"},
"logging": {"request_header_on_debug": "True",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"server": {"script_name": script_name},
"rights": {"type": "owner_only"}})
json_dict: dict
path_base = "/owner/calendar.ics/"
event = get_file_content("event1.ics")
path = path_base + "/event1.ics"
logging.info("\n*** prepare")
self.mkcalendar(path_base, login="owner:ownerpw")
self.put(path, event, login="owner:ownerpw")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
logging.info("\n*** test access to collection")
_, headers, answer = self.request("GET", path_base, check=200, login="owner:ownerpw")
assert "UID:event" in answer
logging.info("\n*** test access to item")
_, headers, answer = self.request("GET", path, check=200, login="owner:ownerpw")
assert "UID:event" in answer
logging.info("\n*** create token")
json_dict = {}
json_dict["PathMapped"] = script_name + path_base
json_dict["Enabled"] = True
json_dict["Hidden"] = False
_, headers, answer = self._sharing_api_json("token", "create", check=200, login="owner:ownerpw", json_dict=json_dict, prefix=script_name)
answer_dict = json.loads(answer)
assert "Status" in answer_dict
assert "PathOrToken" in answer_dict
Token = answer_dict["PathOrToken"]
logging.debug("Token: %r", Token)
assert Token.startswith(script_name) is True
path_shared = Token
# check PROPFIND item as owner (remove prefix again as added later)
logging.info("\n*** PROPFIND item as owner -> calendar")
response = self._propfind_allprop(path_shared.removeprefix(script_name), login="owner:ownerpw", prefix=script_name)
logging.debug("response: %r", response)
assert "CR:supported-address-data" not in response
assert "C:supported-calendar-component-set" in response
assert "D:current-user-privilege-set" in response
def test_sharing_api_token_usage(self) -> None:
"""share-by-token API tests - real usage."""
self.configure({"auth": {"type": "htpasswd",
@@ -703,7 +766,6 @@ class TestSharingApiSanity(BaseTest):
form_array: Sequence[str]
json_dict: dict
path_token = "/.token/"
path_base = "/owner/calendar.ics/"
path_base2 = "/owner/calendar2.ics/"
@@ -761,10 +823,10 @@ class TestSharingApiSanity(BaseTest):
assert "Status='success'" in answer
logging.info("\n*** fetch collection using invalid token (without credentials)")
_, headers, answer = self.request("GET", path_token + "v1/invalidtoken", check=401)
_, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=401)
logging.info("\n*** fetch collection using token (without credentials)")
_, headers, answer = self.request("GET", path_token + token, check=200)
_, headers, answer = self.request("GET", token, check=200)
assert "UID:event" in answer
logging.info("\n*** disable token (form->text)")
@@ -773,7 +835,7 @@ class TestSharingApiSanity(BaseTest):
assert "Status='success'" in answer
logging.info("\n*** fetch collection using disabled token (without credentials)")
_, headers, answer = self.request("GET", path_token + token, check=401)
_, headers, answer = self.request("GET", token, check=401)
logging.info("\n*** enable token (form->text)")
form_array = ["PathOrToken=" + token]
@@ -781,7 +843,7 @@ class TestSharingApiSanity(BaseTest):
assert "Status='success'" in answer
logging.info("\n*** fetch collection using token (without credentials)")
_, headers, answer = self.request("GET", path_token + token, check=200)
_, headers, answer = self.request("GET", token, check=200)
assert "UID:event" in answer
logging.info("\n*** delete token#2 (json->json)")
@@ -804,7 +866,7 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self._sharing_api_form("token", "delete", check=404, login="owner:ownerpw", form_array=form_array)
logging.info("\n*** fetch collection using deleted token (without credentials)")
_, headers, answer = self.request("GET", path_token + token, check=401)
_, headers, answer = self.request("GET", token, check=401)
def test_sharing_api_map_basic(self) -> None:
"""share-by-map API basic tests."""
@@ -4192,17 +4254,17 @@ permissions: RrWw""")
assert 'ICAL:calendar-color' not in answer_dict['Content'][0]['Properties']
assert 'C:calendar-description' not in answer_dict['Content'][0]['Properties']
def test_sharing_api_bday_basic(self) -> None:
"""share-by-bday basic tests."""
def test_sharing_api_map_vcf_bday_basic(self) -> None:
"""share-by-map with conversion=bday basic tests."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_bday": True,
"permit_create_map": True,
"permit_properties_overlay": "True",
"enforce_properties_overlay": "True",
"collection_by_bday": "True"},
"collection_by_map": "True"},
"logging": {"request_header_on_debug": "False",
"response_content_on_debug": "True",
"response_header_on_debug": "True",
@@ -4254,27 +4316,54 @@ permissions: RrWw""")
assert 'Content-Disposition' in headers
assert 'Address%20book.vcf' in headers['Content-Disposition']
# create map
logging.info("\n*** create bday user/owner:r -> ok")
# create map with unsupported permissions
logging.info("\n*** create map(bday) user/owner:r -> fail")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "r"
json_dict['Conversion'] = "bday"
json_dict['Permissions'] = "rw"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
json_dict['Properties'] = {"D:displayname": "Test-BDAY"}
_, headers, answer = self._sharing_api_json("bday", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "create", check=405, login="owner:ownerpw", json_dict=json_dict)
# create map
logging.info("\n*** create map(bday) user/owner:r -> ok")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Conversion'] = "bday"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
json_dict['Properties'] = {"D:displayname": "Test-BDAY"}
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
# enable map by user
logging.info("\n*** enable bday by user")
logging.info("\n*** enable map(bday) by user")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
_, headers, answer = self._sharing_api_json("bday", "enable", check=200, login="user:userpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
# list by user
logging.info("\n*** list by user")
json_dict = {}
_, headers, answer = self._sharing_api_json("map", "list", check=200, login="user:userpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
assert answer_dict['Lines'] == 1
row = answer_dict['Content'][0]
if "SHARING_NO_LEGACY" not in os.environ: # TODO: remove/3.7.0-final
assert row['ShareType'] == "bday" # TODO: remove/3.7.0-final
else: # TODO: remove/3.7.0-final
assert row['ShareType'] == "map"
assert row['Conversion'] == "bday"
# check PROPFIND item as user
logging.info("\n*** PROPFIND item as user -> calendar")
@@ -4384,19 +4473,17 @@ permissions: RrWw""")
status, prop = response["D:getcontenttype"]
assert "text/calendar" in str(prop.text)
def test_sharing_api_bday_complex(self) -> None:
"""share-by-bday complex tests."""
def test_sharing_api_map_vcf_bday_complex(self) -> None:
"""share-by-map with conversion=bday complex tests."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_bday": True,
"permit_create_map": True,
"permit_properties_overlay": "True",
"enforce_properties_overlay": "True",
"collection_by_map": "True",
"collection_by_bday": "True"},
"collection_by_map": "True"},
"logging": {"request_header_on_debug": "False",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
@@ -4447,33 +4534,33 @@ permissions: RrWw""")
assert "contact2" in answer
# create bday
logging.info("\n*** create bday user/owner:r -> ok")
logging.info("\n*** create map(bday) user/owner:r -> ok")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_bday
json_dict['Permissions'] = "r"
json_dict['Conversion'] = "bday"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("bday", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
# enable bday by user
logging.info("\n*** enable bday by user")
logging.info("\n*** enable map(bday) by user")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_bday
_, headers, answer = self._sharing_api_json("bday", "enable", check=200, login="user:userpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
# unhide bday by user
logging.info("\n*** unhide bday by user")
logging.info("\n*** unhide map(bday) by user")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_bday
_, headers, answer = self._sharing_api_json("bday", "unhide", check=200, login="user:userpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "unhide", check=200, login="user:userpw", json_dict=json_dict)
# create map
logging.info("\n*** create map user/owner:r -> ok")
@@ -4522,7 +4609,7 @@ permissions: RrWw""")
<getcontentlength />
</prop>
</propfind>""", login="user:userpw", HTTP_DEPTH="1")
# logging.debug("responses: %r", responses)
logging.debug("responses: %r", responses)
response = responses[path_shared_map]
assert not isinstance(response, int)
logging.debug("response %r: %r", path_shared_map, response)
@@ -4556,17 +4643,17 @@ permissions: RrWw""")
status, prop = response["RADICALE:getcontentcount"]
assert int(str(prop.text)) == 2
def test_sharing_api_bday_self(self) -> None:
"""share-by-bday to self tests."""
def test_sharing_api_map_vcf_bday_self(self) -> None:
"""share-by-map with conversion=bday to self tests."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_bday": True,
"permit_create_map": True,
"permit_properties_overlay": "True",
"enforce_properties_overlay": "True",
"collection_by_bday": "True"},
"collection_by_map": "True"},
"logging": {"request_header_on_debug": "False",
"response_header_on_debug": "True",
"response_content_on_debug": "True",
@@ -4624,9 +4711,10 @@ permissions: RrWw""")
json_dict['User'] = "owner"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
json_dict['Conversion'] = "bday"
json_dict['Enabled'] = False
json_dict['Hidden'] = True
_, headers, answer = self._sharing_api_json("bday", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
# check PROPFIND item as owner
logging.info("\n*** PROPFIND all as owner")
@@ -4659,7 +4747,7 @@ permissions: RrWw""")
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("bday", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
# check PROPFIND item as owner
logging.info("\n*** PROPFIND all as owner")
@@ -4703,3 +4791,155 @@ permissions: RrWw""")
# title from default
assert 'Content-Disposition' in headers
assert 'Calendar.ics' in headers['Content-Disposition']
def test_sharing_api_token_vcf_bday(self) -> None:
"""share-by-bday to a token tests."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_bday": True,
"permit_create_token": True,
"permit_properties_overlay": "True",
"enforce_properties_overlay": "True",
"collection_by_token": "True",
"collection_by_bday": "True"},
"logging": {"request_header_on_debug": "False",
"response_header_on_debug": "True",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"rights": {"type": "owner_only"}})
json_dict: dict
logging.info("\n*** prepare and test access")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
path_mapped = "/owner/adressbook-" + db_type + ".vcf/"
self.create_addressbook(path_mapped, login="owner:ownerpw")
contact = get_file_content("contact1.vcf")
path = path_mapped + "/contact1.vcf"
self.put(path, contact, login="owner:ownerpw")
contact = get_file_content("contact2-with-bday.vcf")
path = path_mapped + "/contact2-with-bday.vcf"
self.put(path, contact, login="owner:ownerpw")
contact = get_file_content("contact3-with-bday.vcf")
path = path_mapped + "/contact3-with-bday.vcf"
self.put(path, contact, login="owner:ownerpw")
# check PROPFIND as owner
logging.info("\n*** PROPFIND collection owner -> ok")
_, responses = self.propfind(path_mapped, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<propname />
</propfind>""", login="owner:ownerpw")
logging.info("response: %r", responses)
response = responses[path_mapped]
assert not isinstance(response, int)
assert "CR:supported-address-data" in response
# execute GET as owner
logging.info("\n*** GET VCF collection owner -> ok")
_, answer = self.get(path_mapped, login="owner:ownerpw")
assert "contact1" in answer
assert "contact2" in answer
assert "NICKNAME-C3" in answer
# create map
logging.info("\n*** create token with bday conversion -> ok")
json_dict = {}
json_dict['User'] = "owner"
json_dict['PathMapped'] = path_mapped
json_dict['Enabled'] = True
json_dict['Hidden'] = False
json_dict['Conversion'] = "bday"
_, headers, answer = self._sharing_api_json("token", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert "Status" in answer_dict
assert "PathOrToken" in answer_dict
Token = answer_dict["PathOrToken"]
path_shared = Token
# execute GET with token
logging.info("\n*** GET bday with token")
_, answer = self.get(path_shared)
assert "VCARD" not in answer
assert "Test-FN-C3 (BDAY)" in answer
assert "Test-FN (BDAY)" in answer
# check PROPFIND item with token
logging.info("\n*** PROPFIND item with token -> calendar")
response = self._propfind_allprop(path_shared)
logging.debug("response: %r", response)
assert "CR:supported-address-data" not in response
assert "D:sync-token" not in response
assert "C:supported-calendar-component-set" in response
assert "D:current-user-privilege-set" in response
# verify content as owner
logging.info("\n*** GET collection owner -> ok")
_, headers, answer = self.request("GET", path_shared)
assert 'Content-Type' in headers
assert 'text/calendar' in headers['Content-Type']
# title from default
assert 'Content-Disposition' in headers
assert 'Calendar.ics' in headers['Content-Disposition']
# create map
logging.info("\n*** create token with bday conversion but unsupported permissions -> fail")
json_dict = {}
json_dict['User'] = "owner"
json_dict['PathMapped'] = path_mapped
json_dict['Enabled'] = True
json_dict['Permissions'] = "rw"
json_dict['Hidden'] = False
json_dict['Conversion'] = "bday"
_, headers, answer = self._sharing_api_json("token", "create", check=405, login="owner:ownerpw", json_dict=json_dict)
def test_sharing_api_token_ics_bday(self) -> None:
"""share-by-token ics with bday conversion (has to fail)."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
"permit_create_bday": True,
"permit_create_token": True,
"permit_properties_overlay": "True",
"enforce_properties_overlay": "True",
"collection_by_token": "True",
"collection_by_bday": "True"},
"logging": {"request_header_on_debug": "False",
"response_header_on_debug": "True",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"rights": {"type": "owner_only"}})
json_dict: dict
logging.info("\n*** prepare and test access")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
path_mapped = "/owner/calendar-" + db_type + ".ics/"
self.mkcalendar(path_mapped, login="owner:ownerpw")
# create map
logging.info("\n*** create token of calendar with bday conversion -> fail")
json_dict = {}
json_dict['User'] = "owner"
json_dict['PathMapped'] = path_mapped
json_dict['Enabled'] = True
json_dict['Hidden'] = False
json_dict['Conversion'] = "bday"
_, headers, answer = self._sharing_api_json("token", "create", check=405, login="owner:ownerpw", json_dict=json_dict)