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

Merge pull request #2015 from pbiering/sharing-review-2

Browse Source 
Sharing review 2
This commit is contained in:
Peter Bieringer
2026-03-07 08:58:25 +01:00
committed by GitHub
parent bc3c2dfd78 86081d7c6a
commit 28f0bf9fa8
16 changed files with 1078 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
DOCUMENTATION.md
View File

@@ -2063,6 +2063,8 @@ Default: 10000
_(>= 3.7.0)_ _(>= 3.7.0)_
See also [Collecition Sharing](https://github.com/Kozea/Radicale/blob/master/SHARING.md).
##### type ##### type
_(>= 3.7.0)_ _(>= 3.7.0)_

499
SHARING.md
View File

@@ -3,7 +3,112 @@
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] 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. With _3.7.0_ major extension was implemented using internal mapping configuration stored in a database and a management API.
## Sharing Implementation
Implemenation of sharing collections is done in case entry exists in sharing database by replacing provided data on request and adjust if required data in responses.
Permissions are filtered by provided `Permissions`.
### CxDAV requests
#### CxDav request "(DELETE|GET|HEAD|PUT)"
* Actions
* map
* Lookup by
* `path` (provided in request)
* `user` (authenticated)
* Replace
* `user` by `Owner`
* `path` by `PathMapped`
* Activate
* `permissions_filter` by `Permissions`
#### CxDav request "REPORT"
* Actions
* map
* back-map response
* Lookup by
* `path` (provided in request)
* `user` (authenticated)
* Replace
* `user` by `Owner`
* `path` by `PathMapped`
* Activate
* `permissions_filter` by `Permissions`
#### CxDav request "PROPFIND" without HTTP_DEPTH=1
* Actions
* map
* back-map response
* overwrite `Properties` if provided
* Lookup by
* `path` (provided in request)
* `user` (authenticated)
* Replace
* `user` by `Owner`
* `path` by `PathMapped`
* Overlay
* `Properties` if provided
* Activate
* `permissions_filter` by `Permissions`
#### CxDav request "PROPFIND" with HTTP_DEPTH=1
* Actions
* extend list
* Lookup for active shares for `user` in sharing database
* Extend list if conditions are met
* `permissions_filter` by `Permissions`
#### CxDav request "PROPPATCH"
* Actions
* map
* adjust properties of a collection
* Lookup by
* `path` (provided in request)
* `user` (authenticated)
* Replace
* `user` by `Owner`
* `path` by `PathMapped`
* Activate
* `permissions_filter` by `Permissions`
* Depending on `permissions_filter`, global options and `Permissions`
* adjust properties of collection
* adjust whitelisted properties in `Properties` for overlay (see OVERLAY_PROPERTIES_WHITELIST)
#### CxDav request "(MKCALENDAR|MKCOL)"
* Action
* check for conflicts
* Lookup by
* `user` (authenticated)
* Verify for non-existence as `PathOrToken` in sharing database
* `path` (provided in request)
#### CxDav request "(MOVE)"
* Action
* map source
* map destination
* Lookup by
* `path` (provided in request)
* `user` (authenticated)
* `to_path` (provided in request)
* `to_user` (same as `user`)
* Replace
* `user` by `Owner` (of `path`)
* `path` by `PathMapped` (of path)
* `to_user` by `Owner` (of `to_path`)
* `to_path` by `PathMapped` (of `to_path`)
* Activate
* `permissions_filter` by `Permissions` (of `to_path`)
* `to_permissions_filter` by `Permissions` (of `to_path`)
## Sharing Configuration Store ## Sharing Configuration Store
@@ -38,24 +143,18 @@ Types of supported sharing configuration:
#### CSV #### CSV
(_>= 3.7.0_)
One CSV file containing one row per sharing config, separated by `;` and containing header with columns from above. One CSV file containing one row per sharing config, separated by `;` and containing header with columns from above.
If given, properties are stored in JSON format in CSV. If given, properties are stored in JSON format in CSV.
#### Files #### 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). 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
### Sharing Access via Maps ### Sharing Access via Maps
(_>= 3.7.0_)
Map-based sharing can be accessed as usual after authentication and authorization. Map-based sharing can be accessed as usual after authentication and authorization.
#### Permission Control #### Permission Control
@@ -74,15 +173,14 @@ In case share should be visible using PROPFIND
* unhide map as owner (can be combined with "create") * unhide map as owner (can be combined with "create")
* unhide map as user (explicit required to avoid sudden visible share) * unhide map as user (explicit required to avoid sudden visible share)
### Sharing Access via Tokens ### Sharing Access via Tokens
(_>= 3.7.0_)
Token-based sharing can be accessed after retrieving the token via Token-based sharing can be accessed after retrieving the token via
Token-URI: `/.token/<Token>` Token-URI: `/.token/<Token>`
Note: requests to not enabled or not even defined tokens will resul tin _401 Not Authorized_
#### Permission Control #### Permission Control
* `permit_create_token` * `permit_create_token`
@@ -98,8 +196,6 @@ Token-URI: `/.token/<Token>`
### Sharing Configuration Management API version 1 ### Sharing Configuration Management API version 1
(_>= 3.7.0_)
Type: POST API Type: POST API
Base-URI: `/.sharing/v1/<ShareType>/<Hook>` Base-URI: `/.sharing/v1/<ShareType>/<Hook>`
@@ -112,16 +208,16 @@ See also test cases in `radicale/tests/test_sharing.py`
Parsing be controlled by `CONTENT_TYPE` Parsing be controlled by `CONTENT_TYPE`
* application/x-www-form-urlencoded (_>= 3.7.0_) * application/x-www-form-urlencoded
* application/json (_>= 3.7.0_) * application/json
##### Output Data Format ##### Output Data Format
Can be selected by `HTTP_ACCEPT` Can be selected by `HTTP_ACCEPT` - default is equal to provided `CONTENT_TYPE`
* text/plain (_>= 3.7.0_) * text/plain
* text/csv (_>= 3.7.0_) - only for "list" * text/csv (only for "list")
* application/json (_>= 3.7.0_) * application/json
##### Accepted Input Data Fields ##### Accepted Input Data Fields
@@ -140,21 +236,26 @@ Can be selected by `HTTP_ACCEPT`
Shows what kind of ShareTypes are supported Shows what kind of ShareTypes are supported
* Example: TEXT * Output: text/plain|application/json
``` * Examples
* form->text
```bash
curl -u user:pass -H "accept: text/plain" -d "" http://localhost:5232/.sharing/v1/all/info curl -u user:pass -H "accept: text/plain" -d "" http://localhost:5232/.sharing/v1/all/info
ApiVersion=1 ApiVersion=1
Status=success Status='success'
FeatureEnabledCollectionByMap=True FeatureEnabledCollectionByMap=True
PermittedCreateCollectionByMap=True PermittedCreateCollectionByMap=True
FeatureEnabledCollectionByToken=True FeatureEnabledCollectionByToken=True
PermittedCreateCollectionByToken=True PermittedCreateCollectionByToken=True
``` ```
* Example: JSON * json->json, parsed with `jq`
``` ```
bash
curl -u user:pass --silent -H "accept: application/json" -d "" http://localhost:5232/.sharing/v1/all/info | jq curl -u user:pass --silent -H "accept: application/json" -d "" http://localhost:5232/.sharing/v1/all/info | jq
{ {
"ApiVersion": 1, "ApiVersion": 1,
@@ -166,46 +267,53 @@ curl -u user:pass --silent -H "accept: application/json" -d "" http://localhost:
} }
``` ```
##### API Hook "(token|map)/create" ##### API Hook "(token|map)/create"
* Authorization * Authorization
* Authenticated user is `Owner`
Authenticated user is `Owner`
###### API Hook "token/create" ###### API Hook "token/create"
Create a share by mapping a collection of an `Owner` to a token. Create a share by mapping a collection of an `Owner` to a token.
* Authorization * Authorization
* `PathMapped` is existing and a collection
Authenticated user as `Owner` has at least read access to `PathMapped` * Authenticated user as `Owner` has at least read access to `PathMapped`
* Global permitted by `permit_create_token = True` or `rights` permission `t`
* Global denied by `permit_create_token = False` or `rights` permission `T`
* Input * Input
| Parameter | Owner | | Parameter | Type | Requirement |
| - | - | - |
| PathMapped | str | mandatory |
| User | str | optional(default:owner) |
| Permissions | str | optional(default:r) |
| Enabled | bool | optional(owner/default:False) |
| Hidden | bool | optional(owner/default:True) |
| Properties | str | optional |
* Output: text/plain|application/json
| Parameter | Type | Value |
| - | - | | - | - |
| Owner | implicit(by authentication) | | PathOrToken | str | (autogenerated token) |
| PathMapped | mandatory |
| User | optional(default:owner) |
| Permissions | optional(default:r) |
| Enabled | optional(owner) |
| Hidden | optional(owner) |
| Properties | optional |
* Output * Examples:
* form->text
| Parameter | Value | ```bash
| - | - | curl -u user:pass -d "PathMapped=/user/testcalendar1/" -d "Enabled=True" -d "Hidden=False" http://localhost:5232/.sharing/v1/token/create
| PathOrToken | (autogenerated token) |
* Example: TEXT
```
curl -u user:pass -d "PathMapped=/user/testcalendar1/" http://localhost:5232/.sharing/v1/token/create
ApiVersion=1 ApiVersion=1
Status=success Status='success'
PathOrToken=v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw= PathOrToken='v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw='
```
* json->json
```bash
curl -u user:pass -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="}
``` ```
###### API Hook "map/create" ###### API Hook "map/create"
@@ -213,60 +321,120 @@ PathOrToken=v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw=
Create a share by mapping a collection of an `Owner` to an `User`. Create a share by mapping a collection of an `Owner` to an `User`.
* Authorization * Authorization
* `PathMapped` is existing and a collection
Authenticated user as `Owner` has at least read access to `PathMapped` * Authenticated user as `Owner` has at least read access to `PathMapped`
* Provided `User` has at least read access to `PathOrToken`
Provided `User` has at least read access to `PathOrToken` * Global permitted by `permit_create_map = True` or `rights` permission `m`
* Global denied by `permit_create_map = False` or `rights` permission `M`
* Input * Input
| Parameter | Value | | Parameter | Type | Requirement |
| - | - | | - | - |
| Owner | implicit(by authentication) | | PathOrToken | str | mandatory |
| PathOrToken | mandatory | | PathMapped | str | mandatory |
| PathMapped | mandatory | | User | str | mandatory |
| User | mandatory | | Permissions | str | optional(default:r) |
| Permissions | optional(default:r) | | Enabled | bool | optional(owner/default:False) |
| Enabled | optional(owner) | | Hidden | bool | optional(owner/default:True) |
| Hidden | optional(owner) |
| Properties | optional | | Properties | optional |
* Output: result status * Output: text/plain|application/json
* Example: TEXT * Examples:
* form->text
``` ```bash
curl -u owner:pass -d "PathOrToken=/user/cal1-from-owner/" -d "PathMapped=/owner/cal1/" -d "User=user" http://localhost:5232/.sharing/v1/map/create curl -u owner:pass -d "PathOrToken=/user/cal1-from-owner/" -d "PathMapped=/owner/testcalendar1/" -d "User=user" -d "Enabled=True" -d "Hidden=False" http://localhost:5232/.sharing/v1/map/create
ApiVersion=1 ApiVersion=1
Status=success Status='success'
``` ```
* json->json
```bash
curl -u owner:pass -H "Content-Type: application/json" -d '{ "PathOrToken": "/user/cal1-from-owner/", "PathMapped": "/owner/testcalendar1/", "User" : "user", "Enabled": true, "Hidden": false}' http://localhost:5232/.sharing/v1/map/create
{"ApiVersion": 1, "Status": "success"}
```
##### API Hook "(map|token|all)/list" ##### API Hook "(map|token|all)/list"
List shares (optional with filter) either owned or assigned as user. List shares (optional with filter) either owned or assigned as user.
* Authorization * Authorization
* Authenticated user as `Owner` or `User`
Authenticated user as `Owner` or `User`
* Input * Input
| Parameter | Filter | | Parameter | Type | Used for |
| - | - | | - | - | - |
| Owner | implicit(by authentication) | | PathOrToken | str | optional |
| User | implicit(by authentication) | | PathMapped | str | optional |
| PathOrToken | optional |
| PathMapped | optional |
* Output: plain/csv/json * Output: text/plain|text/csv|application/json
* Example: CSV * Examples
* form->text ("all")
```bash
curl -u user:pass -d "" http://localhost:5232/.sharing/v1/map/list://localhost:5232/.sharing/v1/map/list
ApiVersion=1
Lines=1
Status='success'
Fields="ShareType;PathOrToken;PathMapped;Owner;User;Permissions;EnabledByOwner;EnabledByUser;HiddenByOwner;HiddenByUser;TimestampCreated;TimestampUpdated;Properties"
Content[0]="map;/user/cal1-from-owner/;/owner/testcalendar1/;owner;user;r;True;True;False;False;1772748001;1772748163;
``` ```
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 * form->csv ("map" only)
map,/user/cal1-from-owner/,/owner/cal1/,owner,user,r,False,False,True,True,1771962120,1771962120
```bash
curl -H "accept: text/csv" -u user:pass -d "" http://localhost:5232/.sharing/v1/map/list://localhost:5232/.sharing/v1/map/list
ShareType;PathOrToken;PathMapped;Owner;User;Permissions;EnabledByOwner;EnabledByUser;HiddenByOwner;HiddenByUser;TimestampCreated;TimestampUpdated;Properties
map;/user/cal1-from-owner/;/owner/testcalendar1/;owner;user;r;True;False;False;True;1772747277;1772747277;
```
* json->json ("all"), parsed with `jq`
```bash
curl -s -H "Content-Type: application/json" -u user:pass -d "{}" http://localhost:5232/.sharing/v1/all/list | jq
{
"ApiVersion": 1,
"Lines": 2,
"Status": "success",
"Content": [
{
"ShareType": "map",
"PathOrToken": "/user/cal1-from-owner/",
"PathMapped": "/owner/testcalendar1/",
"Owner": "owner",
"User": "user",
"Permissions": "r",
"EnabledByOwner": true,
"EnabledByUser": false,
"HiddenByOwner": false,
"HiddenByUser": true,
"TimestampCreated": 1772747277,
"TimestampUpdated": 1772747277,
"Properties": ""
},
{
"ShareType": "token",
"PathOrToken": "v1/DUSl_J5rRlWx3fy8YRXpH22FFllplkOTpcSwfGtpvkc=",
"PathMapped": "/user/testcalendar1/",
"Owner": "user",
"User": "user",
"Permissions": "r",
"EnabledByOwner": true,
"EnabledByUser": false,
"HiddenByOwner": false,
"HiddenByUser": true,
"TimestampCreated": 1772747371,
"TimestampUpdated": 1772747371,
"Properties": ""
}
7]
}
``` ```
@@ -275,16 +443,33 @@ map,/user/cal1-from-owner/,/owner/cal1/,owner,user,r,False,False,True,True,17719
Delete a share selected by `PathOrToken`. Delete a share selected by `PathOrToken`.
* Authorization * Authorization
* Authenticated user is `Owner`
Authenticated user is `Owner` * Share is existing and owned
* Input * Input
| Parameter | Type | Owner | User | | Parameter | Type | Used for | as Owner | as User |
| - | - | - | - | | - | - | - | - | - |
| PathOrToken | selector | mandatory | not-permitted | | PathOrToken | str | selection | mandatory | not-permitted |
* Output: result status * Output: text/plain|application/json
* Examples:
* form->text
```bash
curl -u owner:pass -d "PathOrToken=/user/cal1-from-owner/" http://localhost:5232/.sharing/v1/map/delete
ApiVersion=1
Status='success'
```
* json->json
```bash
curl -u user:pass -H "Content-Type: application/json" -d '{ "PathOrToken": "v1/DUSl_J5rRlWx3fy8YRXpH22FFllplkOTpcSwfGtpvkc="}' http://localhost:5232/.sharing/v1/token/delete
{"ApiVersion": 1, "Status": "success"}
```
##### API Hook "(token|map)/update" ##### API Hook "(token|map)/update"
@@ -293,54 +478,68 @@ Update a share selected by `PathOrToken`.
Execute delete+create in case `PathOrToken` needs to be changed. Execute delete+create in case `PathOrToken` needs to be changed.
* Authorization * Authorization
* Authenticated user is `Owner` or `User`
Authenticated user is `Owner` or `User`
* Input * Input
| Parameter | Type | Owner | User | | Parameter | Type | Used for | Owner | User |
| - | - | - | - | | - | - | - | - | - |
| PathOrToken | selector | mandatory | mandatory | | PathOrToken | str | selection | mandatory | mandatory |
| Owner | by authentication | not-permitted | not-permitted | | PathMapped | str | adjust | optional | not-permitted |
| PathMapped | adjustable | optional | not-permitted | | User | str | adjust | optional | not-permitted |
| User | adjustable | optional | not-permitted | | Permissions | str | adjust | optional | not-permitted |
| Permissions | adjustable | optional | not-permitted | | Enabled | bool | adjust | optional(owner) | optional(user) |
| Enabled | adjustable | optional(owner) | optional(user) | | Hidden | bool | adjust | optional(owner) | optional(user) |
| Hidden | adjustable | optional(owner) | optional(user) | | Properties | str | adjust | optional | optional |
| Properties | adjustable | optional | optional |
* Output: result status * Output: text/plain|application/json
* Examples:
* form->text
```bash
curl -u user:pass -d "PathOrToken=/user/cal1-from-owner/" -d "Enabled=True" -d "Hidden=False" http://localhost:5232/.sharing/v1/map/update
ApiVersion=1
Status='success'
```
* json->json
```bash
curl -u user:pass -H "Content-Type: application/json" -d '{ "PathOrToken": "/user/cal1-from-owner/", "Enabled": true, "Hidden": false}' http://localhost:5232/.sharing/v1/map/update
{"ApiVersion": 1, "Status": "success"}
```
##### API Hooks "(map|token)/(enable|disable|hide|unhide)" ##### API Hooks "(map|token)/(enable|disable|hide|unhide)"
Toggle enable|disable|hide|unhide of `Owner` or `User` of a share selected by `PathOrToken` Toggle enable|disable|hide|unhide of `Owner` or `User` of a share selected by `PathOrToken`
* Authorization * Authorization
* Authenticated user is `Owner` or `User`
Authenticated user is `Owner` or `User` * `PathOrToken` is existing and either owned or assigned to user
* Input * Input
| Parameter | Type | Owner | User | | Parameter | Type | Used for | Owner | User |
| - | - | - | - | | - | - | - | - | - |
| PathOrToken | selector | mandatory | mandatory | | PathOrToken | selection | mandatory | mandatory |
* Output: result status * Output: text/plain|application/json
* Example: TEXT (enable) * form->text
```
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) ```bash
curl -u user:pass -d "PathOrToken=/user/cal1-from-owner/" http://localhost:5232/.sharing/v1/map/enable
```
curl -u owner:pass -d '{"PathOrToken": "/user/cal1-from-owner/", "PathMapped": "/owner/cal1/", "User": "user"} http://localhost:5232/.sharing/v1/map/unhide
ApiVersion=1 ApiVersion=1
Status=success Status='success'
```bash
* json->json
```bash
curl -u user:pass -H "Content-Type: application/json" -d '{ "PathOrToken": "/user/cal1-from-owner/"}' http://localhost:5232/.sharing/v1/map/unhide
{"ApiVersion": 1, "Status": "success"}
``` ```
## Properties Overlay ## Properties Overlay
@@ -349,10 +548,10 @@ Owner or user can define per share a set of properties to overlay on PROPFIND re
Whitelisted ones are defined in `OVERLAY_PROPERTIES_WHITELIST` in `radicale/sharing/__init__.py`: Whitelisted ones are defined in `OVERLAY_PROPERTIES_WHITELIST` in `radicale/sharing/__init__.py`:
* `C:calendar-description` (_>= 3.7.0_) * `C:calendar-description`
* `ICAL:calendar-color` (_>= 3.7.0_) * `ICAL:calendar-color`
* `CR:addressbook-description` (_>= 3.7.0_) * `CR:addressbook-description`
* `INF:addressbook-color` (_>= 3.7.0_) * `INF:addressbook-color`
### Properties Overlay Control Options ### Properties Overlay Control Options
@@ -360,3 +559,67 @@ Whitelisted ones are defined in `OVERLAY_PROPERTIES_WHITELIST` in `radicale/shar
* supported *share* permissions: `Pp` * supported *share* permissions: `Pp`
* `enforce_properties_overlay` * `enforce_properties_overlay`
* supported *share* permissions: `Ee` * supported *share* permissions: `Ee`
### Properties Overlay Example
#### Requirements
* sharing / permit_properties_overlay = True
#### Test sequence
* Prepare XML statements
```bash
## PROPFIND color
xml_pfc='<?xml version="1.0"?>
<propfind xmlns="DAV:" xmlns:ICAL="http://apple.com/ns/ical/">
<prop>
<ICAL:calendar-color />
</prop>
</propfind>'
## PROPPATCH color
xml_ppc='<?xml version="1.0"?>
<D:propertyupdate xmlns:D="DAV:">
<D:set>
<D:prop>
<I:calendar-color xmlns:I="http://apple.com/ns/ical/">#DDDDDD</I:calendar-color>
</D:prop>
</D:set>
</D:propertyupdate>'
```
* Tests
```bash
## Retrieve collection color of owner (no color set)
curl -u owner:pass -d "$xml_pfc" -X PROPFIND http://localhost:5232/owner/testcalendar1/
## Create read-only share for user
curl -u owner:pass -d "PathOrToken=/user/cal1-from-owner/" -d "PathMapped=/owner/testcalendar1/" -d "User=user" -d "Enabled=True" -d "Hidden=False" http://localhost:5232/.sharing/v1/map/create
## Accept (enable+unhide) share by user
curl -u user:pass -d "PathOrToken=/user/cal1-from-owner/" -d "Enabled=True" -d "Hidden=False" http://localhost:5232/.sharing/v1/map/update
## Retrieve collection color of share by user (no color set)
curl -u user:pass -d "$xml_pfc" -X PROPFIND http://localhost:5232/user/cal1-from-owner/
## Set property overlay by user
curl -u user:pass -d "PathOrToken=/user/cal1-from-owner/" -d 'Properties="ICAL:calendar-color"="#CCCCCC"' http://localhost:5232/.sharing/v1/map/update
## Retrieve collection color of share by user (color set)
curl -u user:pass -d "$xml_pfc" -X PROPFIND http://localhost:5232/user/cal1-from-owner/
## Delete property overlay by user
curl -u user:pass -d "PathOrToken=/user/cal1-from-owner/" -d 'Properties=' http://localhost:5232/.sharing/v1/map/update
## Retrieve collection color of share by user (no color set)
curl -u user:pass -d "$xml_pfc" -X PROPFIND http://localhost:5232/user/cal1-from-owner/
## Add property overlay by user using PROPPATCH
curl -u user:pass -d "$xml_ppc" -X PROPPATCH http://localhost:5232/user/cal1-from-owner/
## Retrieve collection color of share by user (color set)
curl -u user:pass -d "$xml_pfc" -X PROPFIND http://localhost:5232/user/cal1-from-owner/
```

10
radicale/app/delete.py
View File

@@ -60,12 +60,12 @@ class ApplicationPartDelete(ApplicationBase):
permissions_filter = None permissions_filter = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"): if not access.check("w"):
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED

10
radicale/app/get.py
View File

@@ -79,12 +79,12 @@ class ApplicationPartGet(ApplicationBase):
permissions_filter = None permissions_filter = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("r") and "i" not in access.permissions: if not access.check("r") and "i" not in access.permissions:
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED

10
radicale/app/mkcalendar.py
View File

@@ -55,11 +55,11 @@ class ApplicationPartMkcalendar(ApplicationBase):
"Bad MKCALENDAR request on %r: %s", path, e, exc_info=True) "Bad MKCALENDAR request on %r: %s", path, e, exc_info=True)
return httputils.BAD_REQUEST return httputils.BAD_REQUEST
if self._sharing._enabled: if self._sharing._enabled:
# check for shared collections (active or inactive) # check for shared collections (all)
collections_shared_map = self._sharing.sharing_collection_map_list(user, active=False) collections_share_map = self._sharing.sharing_collection_map_list()
if collections_shared_map: if collections_share_map:
for sharing in collections_shared_map: for share in collections_share_map:
if sharing['PathOrToken'] == path: if share['PathOrToken'] == path:
return httputils.CONFLICT return httputils.CONFLICT
# TODO: use this? # TODO: use this?
# timezone = props.get("C:calendar-timezone") # timezone = props.get("C:calendar-timezone")

10
radicale/app/mkcol.py
View File

@@ -62,11 +62,11 @@ class ApplicationPartMkcol(ApplicationBase):
logger.warning("MKCOL request %r (type:%s): %s", path, collection_type, "rejected because of missing rights 'W'") logger.warning("MKCOL request %r (type:%s): %s", path, collection_type, "rejected because of missing rights 'W'")
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if self._sharing._enabled: if self._sharing._enabled:
# check for shared collections (active or inactive) # check for shared collections (all)
collections_shared_map = self._sharing.sharing_collection_map_list(user, active=False) collections_share_map = self._sharing.sharing_collection_map_list()
if collections_shared_map: if collections_share_map:
for sharing in collections_shared_map: for share in collections_share_map:
if sharing['PathOrToken'] == path: if share['PathOrToken'] == path:
return httputils.CONFLICT return httputils.CONFLICT
with self._storage.acquire_lock("w", user, path=path, request="MKCOL"): with self._storage.acquire_lock("w", user, path=path, request="MKCOL"):
item = next(iter(self._storage.discover(path)), None) item = next(iter(self._storage.discover(path)), None)

20
radicale/app/move.py
View File

@@ -72,12 +72,12 @@ class ApplicationPartMove(ApplicationBase):
permissions_filter = None permissions_filter = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"): if not access.check("w"):
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
@@ -89,12 +89,12 @@ class ApplicationPartMove(ApplicationBase):
to_path = to_path[len(base_prefix):] to_path = to_path[len(base_prefix):]
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(to_path, to_user) share = self._sharing.sharing_collection_resolver(to_path, to_user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
to_path = sharing['PathMapped'] to_path = share['PathMapped']
to_user = sharing['Owner'] to_user = share['Owner']
to_permissions_filter = sharing['Permissions'] to_permissions_filter = share['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)
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"): if not to_access.check("w"):

60
radicale/app/propfind.py
View File

@@ -37,7 +37,7 @@ from radicale.log import logger
def xml_propfind(base_prefix: str, path: str, def xml_propfind(base_prefix: str, path: str,
xml_request: Optional[ET.Element], xml_request: Optional[ET.Element],
allowed_items: Iterable[Tuple[types.CollectionOrItem, str]], allowed_items: Iterable[Tuple[types.CollectionOrItem, str]],
user: str, encoding: str, max_resource_size: int, sharing: Union[dict, None] = None) -> Optional[ET.Element]: user: str, encoding: str, max_resource_size: int, share: Union[dict, None] = None) -> Optional[ET.Element]:
"""Read and answer PROPFIND requests. """Read and answer PROPFIND requests.
Read rfc4918-9.1 for info. Read rfc4918-9.1 for info.
@@ -74,7 +74,7 @@ def xml_propfind(base_prefix: str, path: str,
write = permission == "w" write = permission == "w"
multistatus.append(xml_propfind_response( multistatus.append(xml_propfind_response(
base_prefix, path, item, props, user, encoding, write=write, base_prefix, path, item, props, user, encoding, write=write,
allprop=allprop, propname=propname, max_resource_size=max_resource_size, sharing=sharing)) allprop=allprop, propname=propname, max_resource_size=max_resource_size, share=share))
return multistatus return multistatus
@@ -82,7 +82,7 @@ def xml_propfind(base_prefix: str, path: str,
def xml_propfind_response( def xml_propfind_response(
base_prefix: str, path: str, item: types.CollectionOrItem, base_prefix: str, path: str, item: types.CollectionOrItem,
props: Sequence[str], user: str, encoding: str, max_resource_size: int, write: bool = False, props: Sequence[str], user: str, encoding: str, max_resource_size: int, write: bool = False,
propname: bool = False, allprop: bool = False, sharing: Union[dict, None] = None) -> ET.Element: propname: bool = False, allprop: bool = False, share: Union[dict, None] = None) -> ET.Element:
"""Build and return a PROPFIND response.""" """Build and return a PROPFIND response."""
if propname and allprop or (props and (propname or allprop)): if propname and allprop or (props and (propname or allprop)):
raise ValueError("Only use one of props, propname and allprops") raise ValueError("Only use one of props, propname and allprops")
@@ -102,9 +102,9 @@ def xml_propfind_response(
collection.path, item.href)) collection.path, item.href))
response = ET.Element(xmlutils.make_clark("D:response")) response = ET.Element(xmlutils.make_clark("D:response"))
href = ET.Element(xmlutils.make_clark("D:href")) href = ET.Element(xmlutils.make_clark("D:href"))
if sharing: if share:
# backmap # backmap
uri = uri.replace(sharing['PathMapped'], sharing['PathOrToken']) uri = uri.replace(share['PathMapped'], share['PathOrToken'])
href.text = xmlutils.make_href(base_prefix, uri) href.text = xmlutils.make_href(base_prefix, uri)
response.append(href) response.append(href)
@@ -183,9 +183,9 @@ def xml_propfind_response(
is_collection and collection.is_principal): is_collection and collection.is_principal):
child_element = ET.Element(xmlutils.make_clark("D:href")) child_element = ET.Element(xmlutils.make_clark("D:href"))
child_element.text = xmlutils.make_href(base_prefix, path) child_element.text = xmlutils.make_href(base_prefix, path)
if sharing: if share:
# backmap # backmap
child_element.text = child_element.text.replace(sharing['PathMapped'], sharing['PathOrToken']) child_element.text = child_element.text.replace(share['PathMapped'], share['PathOrToken'])
element.append(child_element) element.append(child_element)
elif tag == xmlutils.make_clark("C:supported-calendar-component-set"): elif tag == xmlutils.make_clark("C:supported-calendar-component-set"):
human_tag = xmlutils.make_human_tag(tag) human_tag = xmlutils.make_human_tag(tag)
@@ -221,9 +221,9 @@ def xml_propfind_response(
child_element = ET.Element(xmlutils.make_clark("D:href")) child_element = ET.Element(xmlutils.make_clark("D:href"))
child_element.text = xmlutils.make_href( child_element.text = xmlutils.make_href(
base_prefix, "/%s/" % user) base_prefix, "/%s/" % user)
if sharing: if share:
# backmap # backmap
child_element.text = child_element.text.replace(sharing['Owner'], sharing['User']) child_element.text = child_element.text.replace(share['Owner'], share['User'])
element.append(child_element) element.append(child_element)
else: else:
element.append(ET.Element( element.append(ET.Element(
@@ -344,13 +344,13 @@ def xml_propfind_response(
else: else:
human_tag = xmlutils.make_human_tag(tag) human_tag = xmlutils.make_human_tag(tag)
tag_text = collection.get_meta(human_tag) tag_text = collection.get_meta(human_tag)
if share:
# map/add from overlay
if share['Properties']:
if human_tag in share['Properties']:
if share['Properties'][human_tag] is not None:
tag_text = share['Properties'][human_tag]
if tag_text is not None: if tag_text is not None:
if sharing:
# map from overlay
if sharing['Properties']:
if human_tag in sharing['Properties']:
if sharing['Properties'][human_tag] is not None:
tag_text = sharing['Properties'][human_tag]
element.text = tag_text element.text = tag_text
else: else:
is404 = True is404 = True
@@ -426,15 +426,15 @@ class ApplicationPartPropfind(ApplicationBase):
"""Manage PROPFIND request.""" """Manage PROPFIND request."""
http_depth = environ.get("HTTP_DEPTH", "0") http_depth = environ.get("HTTP_DEPTH", "0")
permissions_filter = None permissions_filter = None
sharing = None share = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("r"): if not access.check("r"):
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
@@ -466,14 +466,14 @@ class ApplicationPartPropfind(ApplicationBase):
if http_depth == "1": if http_depth == "1":
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: get shared collections") logger.debug("TRACE/PROPFIND: get shared collections")
# check for shared collections # check for shared collections related to user, Enabled and not Hidden
collections_shared_map = self._sharing.sharing_collection_map_list(user) collections_share_map = self._sharing.sharing_collection_map_list(User=user, Enabled=True, Hidden=False)
if collections_shared_map: if collections_share_map:
for sharing in collections_shared_map: for share in collections_share_map:
c_share = sharing['PathOrToken'] c_share = share['PathOrToken']
c_path = sharing['PathMapped'] c_path = share['PathMapped']
c_user = sharing['Owner'] c_user = share['Owner']
c_permissions_filter = sharing['Permissions'] c_permissions_filter = share['Permissions']
if logger.isEnabledFor(logging.DEBUG): 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) 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) c_access = Access(self._rights, c_user, c_path, c_permissions_filter)
@@ -490,7 +490,7 @@ class ApplicationPartPropfind(ApplicationBase):
headers = {"DAV": httputils.DAV_HEADERS, headers = {"DAV": httputils.DAV_HEADERS,
"Content-Type": "text/xml; charset=%s" % self._encoding} "Content-Type": "text/xml; charset=%s" % self._encoding}
xml_answer = xml_propfind(base_prefix, path, xml_content, xml_answer = xml_propfind(base_prefix, path, xml_content,
allowed_items, user, self._encoding, max_resource_size=self._max_resource_size, sharing=sharing) allowed_items, user, self._encoding, max_resource_size=self._max_resource_size, share=share)
if xml_answer is None: if xml_answer is None:
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content) return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)

58
radicale/app/proppatch.py
View File

@@ -37,7 +37,7 @@ from radicale.log import logger
def xml_proppatch(base_prefix: str, path: str, def xml_proppatch(base_prefix: str, path: str,
xml_request: Optional[ET.Element], xml_request: Optional[ET.Element],
collection: Union[storage.BaseCollection, None], sharing: Union[dict, None] = None, sharing_overlay: bool = False, _sharing: Union[sharing.BaseSharing, None] = None) -> ET.Element: collection: Union[storage.BaseCollection, None], share: Union[dict, None] = None, share_overlay: bool = False, _sharing: Union[sharing.BaseSharing, None] = None) -> ET.Element:
"""Read and answer PROPPATCH requests. """Read and answer PROPPATCH requests.
Read rfc4918-9.2 for info. Read rfc4918-9.2 for info.
@@ -48,9 +48,9 @@ def xml_proppatch(base_prefix: str, path: str,
multistatus.append(response) multistatus.append(response)
href = ET.Element(xmlutils.make_clark("D:href")) href = ET.Element(xmlutils.make_clark("D:href"))
href.text = xmlutils.make_href(base_prefix, path) href.text = xmlutils.make_href(base_prefix, path)
if sharing: if share:
# backmap # backmap
href.text = href.text.replace(sharing['PathMapped'], sharing['PathOrToken']) href.text = href.text.replace(share['PathMapped'], share['PathOrToken'])
response.append(href) response.append(href)
# Create D:propstat element for props with status 200 OK # Create D:propstat element for props with status 200 OK
propstat = ET.Element(xmlutils.make_clark("D:propstat")) propstat = ET.Element(xmlutils.make_clark("D:propstat"))
@@ -62,28 +62,28 @@ def xml_proppatch(base_prefix: str, path: str,
response.append(propstat) response.append(propstat)
props_with_remove = xmlutils.props_from_request(xml_request) props_with_remove = xmlutils.props_from_request(xml_request)
if sharing and sharing_overlay: if share and share_overlay:
# PROPPATCH overlay adjustment # PROPPATCH overlay adjustment
logger.debug("TRACE/PROPPATCH/xml_proppatch: sharing+sharing_overlay is active: %r", sharing) logger.debug("TRACE/PROPPATCH/xml_proppatch: share+share_overlay is active: %r", share)
if sharing['Properties'] is not None: if share['Properties'] is not None:
all_props_with_remove = cast(Dict[str, Optional[str]], radicale_item.check_and_sanitize_props(sharing['Properties'])) all_props_with_remove = cast(Dict[str, Optional[str]], radicale_item.check_and_sanitize_props(share['Properties']))
else: else:
all_props_with_remove = {} all_props_with_remove = {}
all_props_with_remove.update(props_with_remove) all_props_with_remove.update(props_with_remove)
all_props = radicale_item.check_and_sanitize_props(all_props_with_remove) all_props = radicale_item.check_and_sanitize_props(all_props_with_remove)
logger.debug("TRACE/PROPPATCH/xml_proppatch: sharing+sharing_overlay result: %r", all_props) logger.debug("TRACE/PROPPATCH/xml_proppatch: share+share_overlay result: %r", all_props)
else: else:
if collection is not None: if collection is not None:
# always the case, but makes mypy happy # always the case, but makes mypy happy
all_props_with_remove = cast(Dict[str, Optional[str]], dict(collection.get_meta())) all_props_with_remove = cast(Dict[str, Optional[str]], dict(collection.get_meta()))
all_props_with_remove.update(props_with_remove) all_props_with_remove.update(props_with_remove)
all_props = radicale_item.check_and_sanitize_props(all_props_with_remove) all_props = radicale_item.check_and_sanitize_props(all_props_with_remove)
if sharing and sharing_overlay and _sharing is not None: if share and share_overlay and _sharing is not None:
# _sharing is not None: always the case, but makes mypy happy # _sharing is not None: always the case, but makes mypy happy
_sharing.database_update_sharing(ShareType=sharing['ShareType'], _sharing.sharing_collection_update(ShareType=share['ShareType'],
PathOrToken=sharing['PathOrToken'], PathOrToken=share['PathOrToken'],
OwnerOrUser=sharing['User'], OwnerOrUser=share['User'],
Properties=cast(Dict[str, str], all_props)) Properties=cast(Dict[str, str], all_props))
else: else:
if collection is not None: if collection is not None:
# always the case, but makes mypy happy # always the case, but makes mypy happy
@@ -100,21 +100,21 @@ class ApplicationPartProppatch(ApplicationBase):
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse: path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage PROPPATCH request.""" """Manage PROPPATCH request."""
permissions_filter = None permissions_filter = None
sharing = None share = None
sharing_overlay = False share_overlay = False
path_orig = path path_orig = path
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"): if not access.check("w"):
logger.debug("TRACE/PROPPATCH/xml_proppatch: no write-access: %r", path) logger.debug("TRACE/PROPPATCH/xml_proppatch: no write-access: %r", path)
if sharing: if share:
# no write access -> use properties overlay # no write access -> use properties overlay
if self._sharing.permit_properties_overlay: if self._sharing.permit_properties_overlay:
if permissions_filter is not None and "p" in permissions_filter: if permissions_filter is not None and "p" in permissions_filter:
@@ -122,11 +122,11 @@ class ApplicationPartProppatch(ApplicationBase):
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
else: else:
logger.info("PROPPATCH request on shared %r: no write-permissions, overlay permitted by option", path_orig) logger.info("PROPPATCH request on shared %r: no write-permissions, overlay permitted by option", path_orig)
sharing_overlay = True share_overlay = True
else: else:
if permissions_filter is not None and "P" in permissions_filter: if permissions_filter is not None and "P" in permissions_filter:
logger.info("PROPPATCH request on shared %r: no write-permissions, overlay denied, but granted by permission 'P'", path_orig) logger.info("PROPPATCH request on shared %r: no write-permissions, overlay denied, but granted by permission 'P'", path_orig)
sharing_overlay = True share_overlay = True
else: else:
logger.info("PROPPATCH request on shared %r: no write-permissions and overlay denied by option", path_orig) logger.info("PROPPATCH request on shared %r: no write-permissions and overlay denied by option", path_orig)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
@@ -134,18 +134,18 @@ class ApplicationPartProppatch(ApplicationBase):
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
else: else:
logger.debug("TRACE/PROPPATCH/xml_proppatch: write-access: %r", path) logger.debug("TRACE/PROPPATCH/xml_proppatch: write-access: %r", path)
if sharing: if share:
# write access -> check for enforced properties overlay # write access -> check for enforced properties overlay
logger.debug("TRACE/PROPPATCH/xml_proppatch: write-access/sharing: %r", path_orig) logger.debug("TRACE/PROPPATCH/xml_proppatch: write-access/sharing: %r", path_orig)
if self._sharing.enforce_properties_overlay: if self._sharing.enforce_properties_overlay:
if permissions_filter is not None and "e" in permissions_filter: if permissions_filter is not None and "e" in permissions_filter:
logger.info("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by permission 'e'", path_orig) logger.info("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by permission 'e'", path_orig)
else: else:
sharing_overlay = True share_overlay = True
else: else:
if permissions_filter is not None and "E" in permissions_filter: if permissions_filter is not None and "E" in permissions_filter:
logger.info("PROPPATCH request on shared %r: write-permissions, overlay not enforced, but enforced by permission 'E'", path_orig) logger.info("PROPPATCH request on shared %r: write-permissions, overlay not enforced, but enforced by permission 'E'", path_orig)
sharing_overlay = True share_overlay = True
try: try:
xml_content = self._read_xml_request_body(environ) xml_content = self._read_xml_request_body(environ)
except RuntimeError as e: except RuntimeError as e:
@@ -156,13 +156,13 @@ class ApplicationPartProppatch(ApplicationBase):
logger.debug("Client timed out", exc_info=True) logger.debug("Client timed out", exc_info=True)
return httputils.REQUEST_TIMEOUT return httputils.REQUEST_TIMEOUT
if sharing_overlay: if share_overlay:
# call API function internally and no not trigger any hook # call API function internally and no not trigger any hook
headers = {"DAV": httputils.DAV_HEADERS, headers = {"DAV": httputils.DAV_HEADERS,
"Content-Type": "text/xml; charset=%s" % self._encoding} "Content-Type": "text/xml; charset=%s" % self._encoding}
try: try:
xml_answer = xml_proppatch(base_prefix, path, xml_content, xml_answer = xml_proppatch(base_prefix, path, xml_content,
None, sharing, sharing_overlay, self._sharing) None, share, share_overlay, self._sharing)
if xml_content is not None: if xml_content is not None:
content = DefusedET.tostring( content = DefusedET.tostring(
xml_content, xml_content,
@@ -199,7 +199,7 @@ class ApplicationPartProppatch(ApplicationBase):
"Content-Type": "text/xml; charset=%s" % self._encoding} "Content-Type": "text/xml; charset=%s" % self._encoding}
try: try:
xml_answer = xml_proppatch(base_prefix, path, xml_content, xml_answer = xml_proppatch(base_prefix, path, xml_content,
item, sharing) item, share)
if xml_content is not None: if xml_content is not None:
content = DefusedET.tostring( content = DefusedET.tostring(
xml_content, xml_content,

10
radicale/app/put.py
View File

@@ -184,12 +184,12 @@ class ApplicationPartPut(ApplicationBase):
permissions_filter = None permissions_filter = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("w"): if not access.check("w"):

24
radicale/app/report.py
View File

@@ -151,7 +151,7 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
collection: storage.BaseCollection, encoding: str, collection: storage.BaseCollection, encoding: str,
unlock_storage_fn: Callable[[], None], unlock_storage_fn: Callable[[], None],
max_occurrence: int = 0, user: str = "", remote_addr: str = "", remote_useragent: str = "", max_occurrence: int = 0, user: str = "", remote_addr: str = "", remote_useragent: str = "",
sharing: Union[dict, None] = None) -> Tuple[int, ET.Element]: share: Union[dict, None] = None) -> Tuple[int, ET.Element]:
"""Read and answer REPORT requests that return XML. """Read and answer REPORT requests that return XML.
Read rfc3253-3.6 for info. Read rfc3253-3.6 for info.
@@ -360,7 +360,7 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
if found_props or not_found_props: if found_props or not_found_props:
multistatus.append(xml_item_response( multistatus.append(xml_item_response(
base_prefix, uri, found_props=found_props, base_prefix, uri, found_props=found_props,
not_found_props=not_found_props, found_item=True, sharing=sharing)) not_found_props=not_found_props, found_item=True, share=share))
return client.MULTI_STATUS, multistatus return client.MULTI_STATUS, multistatus
@@ -711,13 +711,13 @@ def _find_overridden(
def xml_item_response(base_prefix: str, href: str, def xml_item_response(base_prefix: str, href: str,
found_props: Sequence[ET.Element] = (), found_props: Sequence[ET.Element] = (),
not_found_props: Sequence[ET.Element] = (), not_found_props: Sequence[ET.Element] = (),
found_item: bool = True, sharing: Union[dict, None] = None) -> ET.Element: found_item: bool = True, share: Union[dict, None] = None) -> ET.Element:
response = ET.Element(xmlutils.make_clark("D:response")) response = ET.Element(xmlutils.make_clark("D:response"))
href_element = ET.Element(xmlutils.make_clark("D:href")) href_element = ET.Element(xmlutils.make_clark("D:href"))
href_element.text = xmlutils.make_href(base_prefix, href) href_element.text = xmlutils.make_href(base_prefix, href)
if sharing: if share:
href_element.text = href_element.text.replace(sharing['PathMapped'], sharing['PathOrToken']) href_element.text = href_element.text.replace(share['PathMapped'], share['PathOrToken'])
response.append(href_element) response.append(href_element)
if found_item: if found_item:
@@ -820,15 +820,15 @@ class ApplicationPartReport(ApplicationBase):
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse: path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
"""Manage REPORT request.""" """Manage REPORT request."""
permissions_filter = None permissions_filter = None
sharing = None share = None
if self._sharing._enabled: if self._sharing._enabled:
# Sharing by token or map (if enabled) # Sharing by token or map (if enabled)
sharing = self._sharing.sharing_collection_resolver(path, user) share = self._sharing.sharing_collection_resolver(path, user)
if sharing: if share:
# overwrite and run through extended permission check # overwrite and run through extended permission check
path = sharing['PathMapped'] path = share['PathMapped']
user = sharing['Owner'] user = share['Owner']
permissions_filter = sharing['Permissions'] permissions_filter = share['Permissions']
access = Access(self._rights, user, path, permissions_filter) access = Access(self._rights, user, path, permissions_filter)
if not access.check("r"): if not access.check("r"):
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
@@ -871,7 +871,7 @@ class ApplicationPartReport(ApplicationBase):
try: try:
status, xml_answer = xml_report( status, xml_answer = xml_report(
base_prefix, path, xml_content, collection, self._encoding, base_prefix, path, xml_content, collection, self._encoding,
lock_stack.close, max_occurrence, user, remote_host, remote_useragent, sharing=sharing) lock_stack.close, max_occurrence, user, remote_host, remote_useragent, share=share)
except ValueError as e: except ValueError as e:
logger.warning( logger.warning(
"Bad REPORT request on %r: %s", path, e, exc_info=True) "Bad REPORT request on %r: %s", path, e, exc_info=True)

288
radicale/sharing/__init__.py
View File

@@ -57,8 +57,6 @@ SHARE_TYPES_V1: Sequence[str] = ('token', 'map')
# map : share by mapping collection of one user to another as virtual # map : share by mapping collection of one user to another as virtual
# all : only supported for "list" and "info" # 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') API_HOOKS_V1: Sequence[str] = ('list', 'create', 'delete', 'update', 'hide', 'unhide', 'enable', 'disable', 'info')
# list : list sharings (optional filtered) # list : list sharings (optional filtered)
# create : create share by token or map # create : create share by token or map
@@ -72,6 +70,24 @@ API_HOOKS_V1: Sequence[str] = ('list', 'create', 'delete', 'update', 'hide', 'un
API_SHARE_TOGGLES_V1: Sequence[str] = ('hide', 'unhide', 'enable', 'disable') API_SHARE_TOGGLES_V1: Sequence[str] = ('hide', 'unhide', 'enable', 'disable')
API_TYPES_V1: dict[str, type] = {
"ApiVersion": int,
"Status": str,
"Lines": int,
"FeatureEnabledCollectionByMap": bool,
"FeatureEnabledCollectionByToken": bool,
"PermittedCreateCollectionByMap": bool,
"PermittedCreateCollectionByToken": bool,
"ShareType": str,
"PathOrToken": str,
"PathMapped:": str,
"Owner": str,
"User": str,
"Permissions": str,
"Enabled": bool,
"Hidden": bool,
"Properties": str}
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 PATH_PATTERN: str = "([a-zA-Z0-9/.\\-]+)" # TODO: extend or find better source
@@ -93,6 +109,7 @@ class BaseSharing:
_enabled: bool = False _enabled: bool = False
default_permissions_create_token: str default_permissions_create_token: str
default_permissions_create_map: str default_permissions_create_map: str
sharing_db_type: str
def __init__(self, configuration: "config.Configuration") -> None: def __init__(self, configuration: "config.Configuration") -> None:
"""Initialize Sharing. """Initialize Sharing.
@@ -123,6 +140,10 @@ class BaseSharing:
logger.info("sharing.permit_properties_overlay: %s", self.permit_properties_overlay) logger.info("sharing.permit_properties_overlay: %s", self.permit_properties_overlay)
logger.info("sharing.enforce_properties_overlay: %s", self.enforce_properties_overlay) logger.info("sharing.enforce_properties_overlay: %s", self.enforce_properties_overlay)
# database tasks
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)): 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") logger.info("sharing disabled as no feature is enabled")
self._enabled = False self._enabled = False
@@ -130,15 +151,17 @@ class BaseSharing:
else: else:
self._enabled = True self._enabled = True
# database tasks if not self._init_db():
self.sharing_db_type = configuration.get("sharing", "type") return
logger.info("sharing.database_type: %s", self.sharing_db_type)
def _init_db(self) -> bool:
"""Initialize Sharing Database
"""
try: try:
if self.database_init() is False: if self.database_init() is False:
logger.info("sharing disabled as no database is active") logger.info("sharing disabled as no database is active")
self._enabled = False self._enabled = False
return return False
except Exception as e: except Exception as e:
logger.error("sharing database cannot be initialized: %r", e) logger.error("sharing database cannot be initialized: %r", e)
exit(1) exit(1)
@@ -147,8 +170,9 @@ class BaseSharing:
logger.info("sharing database info: %r", database_info) logger.info("sharing database info: %r", database_info)
else: else:
logger.info("sharing database info: (not provided)") logger.info("sharing database info: (not provided)")
return True
# overloadable database functions # *** overloadable database functions ***
def database_init(self) -> bool: def database_init(self) -> bool:
""" initialize db """ """ initialize db """
return False return False
@@ -216,10 +240,14 @@ class BaseSharing:
""" delete sharing """ """ delete sharing """
return {"status": "not-implemented"} return {"status": "not-implemented"}
# sharing functions called by request methods # *** functions called by cli ***
def verify(self) -> bool: def verify(self) -> bool:
""" verify database """ """ verify database """
logger.info("sharing database verification begin") logger.info("sharing database verification begin")
if not self._init_db():
return False
logger.info("sharing database verification call: %s", self.sharing_db_type) logger.info("sharing database verification call: %s", self.sharing_db_type)
result = self.database_verify() result = self.database_verify()
if result is not True: if result is not True:
@@ -245,10 +273,42 @@ class BaseSharing:
return False return False
else: else:
pass pass
# TODO: check PathMapped exists
# check PathMapped exists
with self._storage.acquire_lock("r", path=entry['PathMapped']):
item = next(iter(self._storage.discover(entry['PathMapped'])), None)
if not item:
logger.error("PathMapped is not existing: %r", entry['PathMapped'])
return False
else:
logger.debug("PathMapped exists(ok): %r", entry['PathMapped'])
logger.info("sharing database verification content successful") logger.info("sharing database verification content successful")
return True return True
# *** sharing functions called by request methods ***
# list sharings of type "map"
def sharing_collection_map_list(self, User: Union[str, None] = None, Enabled: Union[bool, None] = None, Hidden: Union[bool, None] = None) -> list[dict]:
""" returning dict with shared collections by filter(User/Enabled/Hidden) 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 depending on filter
shared_collection_list = self.database_list_sharing(
ShareType="map",
OwnerOrUser=User,
User=User,
EnabledByOwner=Enabled,
EnabledByUser=Enabled,
HiddenByOwner=Hidden,
HiddenByUser=Hidden)
# final
return shared_collection_list
# resolves a path to a share
def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]: def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if not found""" """ returning dict with PathMapped, Owner, Permissions or None if not found"""
if self.sharing_collection_by_token: if self.sharing_collection_by_token:
@@ -272,38 +332,19 @@ class BaseSharing:
logger.debug("TRACE/sharing/map: not active") logger.debug("TRACE/sharing/map: not active")
return None return None
# final
return None return None
# list sharings of type "map" # adjust a share
def sharing_collection_map_list(self, user: str, active: bool = True) -> list[dict]: def sharing_collection_update(self, ShareType: str, PathOrToken: str, OwnerOrUser: str, Properties: dict) -> None:
""" returning dict with shared collections (active==True: enabled and unhidden) or None if not found""" """ returning dict with PathMapped, Owner, Permissions or None if not found"""
if not self.sharing_collection_by_map: logger.info("Sharing/collection/update: ShareType=%r PathOrToken=%r OwnerOrUser=%r", ShareType, PathOrToken, OwnerOrUser)
if logger.isEnabledFor(logging.DEBUG): self.database_update_sharing(ShareType=ShareType,
logger.debug("TRACE/sharing/map: not active") PathOrToken=PathOrToken,
return [{}] OwnerOrUser=OwnerOrUser,
Properties=Properties)
# retrieve collections which are enabled and not hidden by owner+user # *** internal sharing functions ***
if active: # resolves a token "path" to a share
shared_collection_list = self.database_list_sharing(
ShareType="map",
OwnerOrUser=user,
User=user,
EnabledByOwner=True,
EnabledByUser=True,
HiddenByOwner=False,
HiddenByUser=False)
else:
# unconditional
shared_collection_list = self.database_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]: def sharing_collection_by_token_resolver(self, path) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if invalid""" """ returning dict with PathMapped, Owner, Permissions or None if invalid"""
if self.sharing_collection_by_token: if self.sharing_collection_by_token:
@@ -320,9 +361,12 @@ class BaseSharing:
# TODO add token validity checks # TODO add token validity checks
if logger.isEnabledFor(logging.DEBUG): 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: supported token found in path: %r (token=%r)", path, match[1])
return self.database_get_sharing( result = self.database_get_sharing(
ShareType="token", ShareType="token",
PathOrToken=match[1]) 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'])
return result
else: else:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: no supported prefix found in path: %r", path) logger.debug("TRACE/sharing/token: no supported prefix found in path: %r", path)
@@ -332,6 +376,7 @@ class BaseSharing:
logger.debug("TRACE/sharing/token: not active") logger.debug("TRACE/sharing/token: not active")
return None return None
# resolves a map "path" to a share
def sharing_collection_by_map_resolver(self, path: str, user: str) -> Union[dict, 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""" """ returning dict with PathMapped, Owner, Permissions or None if invalid"""
if self.sharing_collection_by_map: if self.sharing_collection_by_map:
@@ -342,7 +387,7 @@ class BaseSharing:
PathOrToken=path, PathOrToken=path,
User=user) User=user)
if result: if result:
return result pass
else: else:
# fallback to parent path # fallback to parent path
parent_path = pathutils.parent_path(path) parent_path = pathutils.parent_path(path)
@@ -356,17 +401,19 @@ class BaseSharing:
result['PathMapped'] = path.replace(parent_path, result['PathMapped']) result['PathMapped'] = path.replace(parent_path, result['PathMapped'])
if logger.isEnabledFor(logging.DEBUG): 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) logger.debug("TRACE/sharing/map/resolver: PathMapped=%r Permissions=%r by parent_path=%r", result['PathMapped'], result['Permissions'], parent_path)
return result
else: else:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not found") logger.debug("TRACE/sharing/map: not found")
return None 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
else: else:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active") logger.debug("TRACE/sharing/map: not active")
return None return None
# POST API # *** POST API ***
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse: def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse:
# Late import to avoid circular dependency in config # Late import to avoid circular dependency in config
from radicale.app.base import Access from radicale.app.base import Access
@@ -469,11 +516,14 @@ class BaseSharing:
logger.debug("Client timed out", exc_info=True) logger.debug("Client timed out", exc_info=True)
return httputils.REQUEST_TIMEOUT return httputils.REQUEST_TIMEOUT
api_info = "sharing/API/POST/" + ShareType + "/" + action # initial log prefix
api_info = "Sharing/API/POST/" + ShareType + "/" + action
# parse body according to content-type # parse body according to content-type
content_type = environ.get("CONTENT_TYPE", "") content_type = environ.get("CONTENT_TYPE", "")
if 'application/json' in content_type: if 'application/json' in content_type:
input_format = "json"
output_format = "json" # default
try: try:
request_data = json.loads(request_body) request_data = json.loads(request_body)
except json.JSONDecodeError: except json.JSONDecodeError:
@@ -482,12 +532,14 @@ class BaseSharing:
# convert JSON boolean # convert JSON boolean
if key in request_data: if key in request_data:
if type(request_data[key]) is not bool: if type(request_data[key]) is not bool:
logger.error(api_info + ": unsupported (non-boolean) " + key + ": " + request_data[key]) logger.warning(api_info + ": unsupported (non-boolean) " + key + ": " + request_data[key])
return httputils.bad_request("Invalid non-boolean value for " + key + ": " + request_data[key]) return httputils.bad_request("Invalid non-boolean value for " + key + ": " + request_data[key])
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + " (json): %r", f"{request_data}") logger.debug("TRACE/" + api_info + " (json): %r", f"{request_data}")
elif 'application/x-www-form-urlencoded' in content_type: elif 'application/x-www-form-urlencoded' in content_type:
request_parsed = parse_qs(request_body) input_format = "form"
output_format = "plain" # default
request_parsed = parse_qs(request_body, keep_blank_values=True)
# convert arrays into single value # convert arrays into single value
request_data = {} request_data = {}
for key in request_parsed: for key in request_parsed:
@@ -495,6 +547,10 @@ class BaseSharing:
# Properties key value parser # Properties key value parser
properties_dict: dict = {} properties_dict: dict = {}
for entry in request_parsed[key]: for entry in request_parsed[key]:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: parse property %r", entry)
if entry == "":
continue
m = re.search('^([^=]+)=([^=]+)$', entry) m = re.search('^([^=]+)=([^=]+)$', entry)
if not m: if not m:
return httputils.bad_request("Invalid properties format in form") return httputils.bad_request("Invalid properties format in form")
@@ -504,11 +560,14 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: converted Properties from form into dict: %r", properties_dict) logger.debug("TRACE/sharing/API: converted Properties from form into dict: %r", properties_dict)
request_data[key] = properties_dict request_data[key] = properties_dict
if len(request_data[key]) == 0:
# empty
request_data[key] = {}
elif key in ["Enabled", "Hidden"]: elif key in ["Enabled", "Hidden"]:
try: try:
request_data[key] = config._convert_to_bool(request_parsed[key][0]) request_data[key] = config._convert_to_bool(request_parsed[key][0])
except ValueError: except ValueError:
logger.error(api_info + ": unsupported (non-boolean) " + key + ": " + request_parsed[key][0]) logger.warning(api_info + ": unsupported (non-boolean) " + key + ": " + request_parsed[key][0])
return httputils.bad_request("Invalid non-boolean value for " + key + ": " + request_parsed[key][0]) return httputils.bad_request("Invalid non-boolean value for " + key + ": " + request_parsed[key][0])
else: else:
request_data[key] = request_parsed[key][0] request_data[key] = request_parsed[key][0]
@@ -525,23 +584,28 @@ class BaseSharing:
output_format = "json" output_format = "json"
elif 'text/csv' in accept: elif 'text/csv' in accept:
output_format = "csv" output_format = "csv"
elif 'text/plain' in accept:
output_format = "plain"
else: else:
output_format = "txt" # default from input type
pass
if output_format == "csv": if output_format == "csv":
if not action == "list": if not action == "list":
return httputils.bad_request("CSV output format is only allowed for list action") return httputils.bad_request("CSV output format is only allowed for list action")
elif output_format == "json": elif output_format == "json":
pass pass
elif output_format == "txt": elif output_format == "plain":
pass pass
else: else:
return httputils.bad_request("Output format not supported") return httputils.bad_request("Output format not supported")
# extend log prefix
api_info = api_info + "(" + input_format + "->" + output_format + ")"
# parameters default # parameters default
PathOrToken: Union[str, None] = None PathOrToken: Union[str, None] = None
PathMapped: Union[str, None] = None PathMapped: Union[str, None] = None
Owner: str = user
User: Union[str, None] = None User: Union[str, None] = None
Permissions: Union[str, None] = None # no permissions by default Permissions: Union[str, None] = None # no permissions by default
Enabled: Union[bool, None] = None Enabled: Union[bool, None] = None
@@ -557,23 +621,23 @@ class BaseSharing:
elif key == "PathOrToken": elif key == "PathOrToken":
if ShareType == "token": if ShareType == "token":
if not re.search('^' + TOKEN_PATTERN_V1 + '$', request_data[key]): if not re.search('^' + TOKEN_PATTERN_V1 + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key) logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken") return httputils.bad_request("Invalid value for PathOrToken")
elif ShareType == "map": elif ShareType == "map":
if not re.search('^' + PATH_PATTERN + '$', request_data[key]): if not re.search('^' + PATH_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key) logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken") return httputils.bad_request("Invalid value for PathOrToken")
elif not request_data[key].endswith("/"): elif not request_data[key].endswith("/"):
return httputils.bad_request("PathOrToken not ending with /") return httputils.bad_request("PathOrToken not ending with /")
elif key == "PathMapped": elif key == "PathMapped":
if not re.search('^' + PATH_PATTERN + '$', request_data[key]): if not re.search('^' + PATH_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key) logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathMapped") return httputils.bad_request("Invalid value for PathMapped")
elif not request_data[key].endswith("/"): elif not request_data[key].endswith("/"):
return httputils.bad_request("PathMapped not ending with /") return httputils.bad_request("PathMapped not ending with /")
elif key == "User": elif key == "User":
if not re.search('^' + USER_PATTERN + '$', request_data[key]): if not re.search('^' + USER_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key) logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for User") return httputils.bad_request("Invalid value for User")
# check for optional parameters # check for optional parameters
@@ -586,7 +650,7 @@ class BaseSharing:
# ignored # ignored
pass pass
elif action not in ['list', 'create']: elif action not in ['list', 'create']:
logger.error(api_info + ": missing PathOrToken") logger.warning(api_info + ": missing PathOrToken")
return httputils.bad_request("Missing PathOrToken") return httputils.bad_request("Missing PathOrToken")
else: else:
# PathOrToken is optional # PathOrToken is optional
@@ -594,7 +658,7 @@ class BaseSharing:
else: else:
if action == "create" and ShareType == "token": if action == "create" and ShareType == "token":
# not supported # not supported
logger.error(api_info + ": PathOrToken found but not supported") logger.warning(api_info + ": PathOrToken found but not supported")
return httputils.bad_request("PathOrToken not supported") return httputils.bad_request("PathOrToken not supported")
PathOrToken = request_data['PathOrToken'] PathOrToken = request_data['PathOrToken']
@@ -662,15 +726,25 @@ class BaseSharing:
answer['Status'] = "success" answer['Status'] = "success"
answer['Content'] = result_array answer['Content'] = result_array
logger.info(api_info + ": " + answer['Status'])
# action: create # action: create
elif action == "create": elif action == "create":
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start") logger.debug("TRACE/" + api_info + ": start")
if PathMapped is None: if PathMapped is None:
logger.error(api_info + ": missing PathMapped") logger.warning(api_info + ": missing PathMapped")
return httputils.bad_request("Missing PathMapped") return httputils.bad_request("Missing PathMapped")
# check whether collection exists
with self._storage.acquire_lock("r", user, path=PathMapped):
item = next(iter(self._storage.discover(PathMapped)), None)
if not item:
return httputils.NOT_FOUND
if not isinstance(item, storage.BaseCollection):
return httputils.METHOD_NOT_ALLOWED
if Permissions is None: if Permissions is None:
if ShareType == "token": if ShareType == "token":
Permissions = self.default_permissions_create_token Permissions = self.default_permissions_create_token
@@ -701,16 +775,16 @@ class BaseSharing:
# check access Permissions # check access Permissions
access = Access(self._rights, user, PathMapped) access = Access(self._rights, user, PathMapped)
if not access.check("r"): if not access.check("r"):
logger.info("Add sharing-by-token: access to %r not allowed for user %r", PathMapped, user) logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if self.permit_create_token is False: if self.permit_create_token is False:
if "t" not in access.permissions: if "t" not in access.permissions:
logger.info("Add sharing-by-token: access to %r not allowed for user %r (permit=False but explict grant misses 't')", PathMapped, user) logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=False but explict grant misses 't')", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
else: else:
if "T" in access.permissions: if "T" in access.permissions:
logger.info("Add sharing-by-token: access to %r not allowed for user %r (permit=True but denied by 'T')", PathMapped, user) logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'T')", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if User is not None: if User is not None:
@@ -752,7 +826,7 @@ class BaseSharing:
# retrieve existing share # retrieve existing share
share = self.database_get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False) share = self.database_get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False)
if share is not None: if share is not None:
logger.error("Sharing/create/%s: already exists: %r", ShareType, PathOrToken) logger.warning(api_info + ": share already exists PathOrToken=%r", PathOrToken)
return httputils.CONFLICT return httputils.CONFLICT
if User is None: if User is None:
@@ -761,36 +835,36 @@ class BaseSharing:
User = str(User) User = str(User)
# check access Permissions # check access Permissions
access = Access(self._rights, Owner, PathMapped, None) # PathMapped is mandatory access = Access(self._rights, user, PathMapped, None) # PathMapped is mandatory
if not access.check("r") and "i" not in access.permissions: 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) logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if self.permit_create_map is False: if self.permit_create_map is False:
if "m" not in access.permissions: if "m" not in access.permissions:
logger.info("Add sharing-by-map: access to %r not allowed for user %r (permit=False but explicit grant misses 'm')", PathMapped, user) 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 return httputils.NOT_ALLOWED
else: else:
if "M" in access.permissions: if "M" in access.permissions:
logger.info("Add sharing-by-map: access to %r not allowed for user %r (permit=True but denied by 'M')", PathMapped, user) logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=True but denied by 'M')", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
access = Access(self._rights, User, PathOrToken) access = Access(self._rights, User, PathOrToken)
if not access.check("r"): if not access.check("r"):
logger.info("Add sharing-by-map: access to path %r not allowed for user %r", PathOrToken, User) logger.warning(api_info + ": access to PathOrToken=%r not allowed for User=%r", PathOrToken, User)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
# check whether share is already existing as real collection # check whether share is already existing as real collection
with self._storage.acquire_lock("r", user, path=PathOrToken): with self._storage.acquire_lock("r", User, path=PathOrToken):
item = next(iter(self._storage.discover(PathOrToken)), None) item = next(iter(self._storage.discover(PathOrToken)), None)
if not item: if not item:
pass pass
else: else:
logger.info("Add sharing-by-map: path %r already exists as real collection for user %r", PathOrToken, user) logger.warning(api_info + ": PathOrToken=%r already exists as real collection for User=%r", PathOrToken, User)
return httputils.CONFLICT return httputils.CONFLICT
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": %r (Permissions=%r PathOrToken=%r user=%r)", PathMapped, Permissions, PathOrToken, User) logger.debug("TRACE/" + api_info + ": %r (Permissions=%r PathOrToken=%r Owner=%r User=%r)", PathMapped, Permissions, PathOrToken, user, User)
result = self.database_create_sharing( result = self.database_create_sharing(
ShareType=ShareType, ShareType=ShareType,
@@ -807,7 +881,7 @@ class BaseSharing:
Properties=Properties) Properties=Properties)
else: else:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType) logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type") return httputils.bad_request("Invalid share type")
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": result=%r", result) logger.debug("TRACE/" + api_info + ": result=%r", result)
@@ -819,19 +893,22 @@ class BaseSharing:
elif result['status'] == "success": elif result['status'] == "success":
answer['Status'] = "success" answer['Status'] = "success"
else: else:
return httputils.bad_request("Internal failure") logger.warning(api_info + ": %r by user %r not successful", PathMapped, request_data['User'])
return httputils.bad_request("Internal Error")
if ShareType == "token": if ShareType == "token":
logger.info(api_info + "(success): %r (Permissions=%r token=%r)", PathMapped, Permissions, token) PathOrToken = token
answer['PathOrToken'] = token answer['PathOrToken'] = token
logger.info(api_info + " success: PathMapped=%r Permissions=%r PathOrToken=%r", PathMapped, Permissions, PathOrToken)
# action: update # action: update
elif action == "update": elif action == "update":
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start") logger.debug("TRACE/" + api_info + ": start")
if ShareType not in ["token", "map"]: if ShareType not in ["token", "map"]:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType) logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type") return httputils.bad_request("Invalid share type")
if PathOrToken is None: if PathOrToken is None:
@@ -849,9 +926,13 @@ class BaseSharing:
# check access Permissions # check access Permissions
access = Access(self._rights, user, str(PathMapped), None) access = Access(self._rights, user, str(PathMapped), None)
if not access.check("r") and "i" not in access.permissions: if not access.check("r") and "i" not in access.permissions:
logger.warning("Update sharing: access to PathMapped %r not allowed for user %r", PathMapped, user) logger.warning(api_info + ": access to %r not allowed for user %r", PathMapped, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if 'Properties' in request_data and Properties is None:
# clear properties
Properties = {}
result = self.database_update_sharing( result = self.database_update_sharing(
ShareType=ShareType, ShareType=ShareType,
PathMapped=PathMapped, PathMapped=PathMapped,
@@ -867,25 +948,29 @@ class BaseSharing:
elif user == share['User']: elif user == share['User']:
# User is only allowed to update Properties # User is only allowed to update Properties
if PathMapped is not None or Permissions is not None or User is not None: if PathMapped is not None or Permissions is not None or User is not None:
logger.info("Update sharing: access to %r not allowed for user %r to adjust anything beside: %s", PathOrToken, user, " ".join(DB_FIELDS_V1_USER_PERMITTED)) logger.warning(api_info + ": access to %r not allowed for user %r to adjust anything beside: %s", PathOrToken, user, " ".join(DB_FIELDS_V1_USER_PERMITTED))
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if Properties is not None: if 'Properties' in request_data:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/update: permit_properties_overlay=%s Permissions=%r", self.permit_properties_overlay, share['Permissions']) logger.debug("TRACE/sharing/API/update: permit_properties_overlay=%s Permissions=%r", self.permit_properties_overlay, share['Permissions'])
if self.permit_properties_overlay: if self.permit_properties_overlay:
if share['Permissions'] is not None and "p" in str(share['Permissions']): if share['Permissions'] is not None and "p" in str(share['Permissions']):
logger.info("Update on shared %r: overlay permitted, but denied by permission 'p'", PathOrToken) logger.warning(api_info + ": %r properties overlay permitted by option, but denied by permission 'p'", PathOrToken)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
else: else:
logger.info("Update on shared %r: overlay permitted by option", PathOrToken) logger.info(api_info + ": %r properties overlay permitted by option", PathOrToken)
else: else:
if share['Permissions'] is not None and "P" in str(share['Permissions']): if share['Permissions'] is not None and "P" in str(share['Permissions']):
logger.info("Update on shared %r: overlay denied, but granted by permission 'P'", PathOrToken) logger.info(api_info + ": %r properties overlay denied by option, but granted by permission 'P'", PathOrToken)
else: else:
logger.info("Update on shared %r: overlay denied by option", PathOrToken) logger.warning(api_info + ": %r properties overlay denied by option", PathOrToken)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if 'Properties' in request_data and Properties is None:
# clear properties
Properties = {}
# limited update as user # limited update as user
result = self.database_update_sharing( result = self.database_update_sharing(
ShareType=ShareType, ShareType=ShareType,
@@ -897,7 +982,7 @@ class BaseSharing:
else: else:
# neither owner nor user matches # neither owner nor user matches
logger.warning("Update sharing of %r not permitted for user %r", PathOrToken, user) logger.warning(api_info + ": sharing of %r not permitted for user %r", PathOrToken, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
# result handling # result handling
@@ -909,11 +994,8 @@ class BaseSharing:
answer['Status'] = "success" answer['Status'] = "success"
pass pass
else: else:
if ShareType == "token": logger.warning(api_info + ": %r not successful", request_data['PathOrToken'])
logger.info("Update of sharing-by-token: %r not successful", request_data['PathOrToken']) return httputils.bad_request("Internal Error")
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 # action: delete
elif action == "delete": elif action == "delete":
@@ -921,7 +1003,7 @@ class BaseSharing:
logger.debug("TRACE/" + api_info + ": start") logger.debug("TRACE/" + api_info + ": start")
if ShareType not in ["token", "map"]: if ShareType not in ["token", "map"]:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType) logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type") return httputils.bad_request("Invalid share type")
if PathOrToken is None: if PathOrToken is None:
@@ -940,7 +1022,7 @@ class BaseSharing:
PathOrToken=PathOrToken) # verification above that it is not None PathOrToken=PathOrToken) # verification above that it is not None
else: else:
# only owner is permitted to delete a share # only owner is permitted to delete a share
logger.warning("Delete sharing of %r not permitted for user %r", PathOrToken, user) logger.warning(api_info + ": %r not permitted for user %r", PathOrToken, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
# result handling # result handling
@@ -952,14 +1034,12 @@ class BaseSharing:
answer['Status'] = "success" answer['Status'] = "success"
pass pass
else: else:
if ShareType == "token": logger.warning(api_info + ": %r by user %r not successful", request_data['PathOrToken'], request_data['User'])
logger.info("Delete sharing-by-token: %r of user %r not successful", request_data['PathOrToken'], request_data['User']) return httputils.bad_request("Internal Error")
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 # action: info
elif action == "info": elif action == "info":
logger.info(api_info + ": success")
answer['Status'] = "success" answer['Status'] = "success"
if ShareType in ["all", "map"]: if ShareType in ["all", "map"]:
answer['FeatureEnabledCollectionByMap'] = self.sharing_collection_by_map answer['FeatureEnabledCollectionByMap'] = self.sharing_collection_by_map
@@ -974,7 +1054,7 @@ class BaseSharing:
logger.debug("TRACE/sharing/API/POST/" + action) logger.debug("TRACE/sharing/API/POST/" + action)
if ShareType not in ["token", "map"]: if ShareType not in ["token", "map"]:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType) logger.warning(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type") return httputils.bad_request("Invalid share type")
if PathOrToken is None: if PathOrToken is None:
@@ -1027,7 +1107,7 @@ class BaseSharing:
else: else:
# neither owner nor user matches # neither owner nor user matches
logger.warning("Toggle sharing of %r not permitted for user %r", PathOrToken, user) logger.warning(api_info + ": %r by user %r not permitted", PathOrToken, user)
return httputils.NOT_ALLOWED return httputils.NOT_ALLOWED
if result: if result:
@@ -1039,29 +1119,34 @@ class BaseSharing:
answer['Status'] = "success" answer['Status'] = "success"
pass pass
else: else:
logger.error("Toggle sharing: %r of user %s not successful", request_data['PathOrToken'], user) logger.warning(api_info + ": %r by user %s not successful", request_data['PathOrToken'], user)
return httputils.bad_request("Internal Error") return httputils.bad_request("Internal Error")
else: else:
# default # default
logger.error(api_info + ": unsupported action=%r", action) logger.warning(api_info + ": unsupported action=%r", action)
return httputils.bad_request("Invalid action") return httputils.bad_request("Invalid action")
# output handler # output handler
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/POST output format: %r", output_format) logger.debug("TRACE/sharing/API/POST output format: %r", output_format)
logger.debug("TRACE/sharing/API/POST answer: %r", answer) logger.debug("TRACE/sharing/API/POST answer: %r", answer)
if output_format == "csv" or output_format == "txt": if output_format == "csv" or output_format == "plain":
answer_array = [] answer_array = []
if output_format == "txt": if output_format == "plain":
for key in answer: for key in answer:
if key != 'Content': if key != 'Content':
answer_array.append(key + '=' + str(answer[key])) if API_TYPES_V1[key] is bool or API_TYPES_V1[key] is int:
answer_array.append(key + '=' + str(answer[key]))
else:
answer_array.append(key + "='" + str(answer[key]) + "'")
if 'Content' in answer and answer['Content'] is not None: if 'Content' in answer and answer['Content'] is not None:
csv = io.StringIO() csv = io.StringIO()
writer = DictWriter(csv, fieldnames=DB_FIELDS_V1, delimiter=';') writer = DictWriter(csv, fieldnames=DB_FIELDS_V1, delimiter=';')
if output_format == "csv": if output_format == "csv":
writer.writeheader() writer.writeheader()
elif output_format == "plain":
writer.writeheader()
for entry in answer['Content']: for entry in answer['Content']:
# TODO: Argument 1 to "writerow" of "DictWriter" has incompatible type "str"; expected "Mapping[str, Any]" [arg-type] # TODO: Argument 1 to "writerow" of "DictWriter" has incompatible type "str"; expected "Mapping[str, Any]" [arg-type]
writer.writerow(entry) # type: ignore[arg-type] writer.writerow(entry) # type: ignore[arg-type]
@@ -1071,7 +1156,10 @@ class BaseSharing:
index = 0 index = 0
for line in csv.getvalue().splitlines(): for line in csv.getvalue().splitlines():
# create a shell array with content lines # create a shell array with content lines
answer_array.append('Content[' + str(index) + ']="' + line.replace('"', '\\"') + '"') if index == 0:
answer_array.append('Fields="' + line + '"')
else:
answer_array.append('Content[' + str(index - 1) + ']="' + line.replace('"', '\\"') + '"')
index += 1 index += 1
headers = { headers = {
"Content-Type": "text/csv" "Content-Type": "text/csv"

313
radicale/sharing/csv.py
View File

@@ -30,7 +30,7 @@ class Sharing(sharing.BaseSharing):
_sharing_cache: list[dict] = [] _sharing_cache: list[dict] = []
_sharing_db_file: str _sharing_db_file: str
# Overloaded functions # *** Overloaded functions ***
def database_init(self) -> bool: def database_init(self) -> bool:
logger.debug("sharing database initialization for type 'csv'") logger.debug("sharing database initialization for type 'csv'")
sharing_db_file = self.configuration.get("sharing", "database_path") sharing_db_file = self.configuration.get("sharing", "database_path")
@@ -158,48 +158,49 @@ class Sharing(sharing.BaseSharing):
index = 0 index = 0
result = [] result = []
if logger.isEnabledFor(logging.DEBUG): with self._storage.acquire_lock("r", path=self._sharing_db_file):
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) 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)
for row in self._sharing_cache: for row in self._sharing_cache:
if index == 0: if index == 0:
# skip fieldnames # 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 pass
else: else:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/list/row: add : %r", row) logger.debug("TRACE/sharing/list/row: test: %r", row)
result.append(row) if ShareType is not None and row['ShareType'] != ShareType:
index += 1 if logger.isEnabledFor(logging.DEBUG):
return result 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 database_create_sharing(self, def database_create_sharing(self,
ShareType: str, ShareType: str,
@@ -213,54 +214,55 @@ class Sharing(sharing.BaseSharing):
""" create sharing """ """ create sharing """
row: dict row: dict
if logger.isEnabledFor(logging.DEBUG): with self._storage.acquire_lock("w", path=self._sharing_db_file):
logger.debug("TRACE/sharing: ShareType=%r", ShareType)
if ShareType == "token":
if logger.isEnabledFor(logging.DEBUG): 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) logger.debug("TRACE/sharing: ShareType=%r", ShareType)
# check for duplicate token entry if ShareType == "token":
for row in self._sharing_cache: if logger.isEnabledFor(logging.DEBUG):
if row['ShareType'] != "token": logger.debug("TRACE/sharing/token/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", PathOrToken, Owner, PathMapped, User, Permissions)
continue # check for duplicate token entry
if row['PathOrToken'] == PathOrToken: for row in self._sharing_cache:
# must be unique systemwide if row['ShareType'] != "token":
logger.error("sharing/token/create: PathOrToken already exists: PathOrToken=%r", PathOrToken) continue
return {"status": "conflict"} if row['PathOrToken'] == PathOrToken:
elif ShareType == "map": # 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): 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) logger.debug("TRACE/sharing/*/create: add row: %r", row)
# check for duplicate map entry self._sharing_cache.append(row)
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 self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: write CSV done", ShareType) logger.debug("TRACE/sharing/%s/create: write CSV done", ShareType)
return {"status": "success"} return {"status": "success"}
logger.error("sharing/%s/create: cannot update CSV database", ShareType) logger.error("sharing/%s/create: cannot update CSV database", ShareType)
return {"status": "error"} return {"status": "error"}
@@ -281,61 +283,59 @@ class Sharing(sharing.BaseSharing):
if logger.isEnabledFor(logging.DEBUG): 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) 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)
# lookup token with self._storage.acquire_lock("w", path=self._sharing_db_file):
found = False # lookup token
index = 0 found = False
for row in self._sharing_cache: index = 0
if index == 0: for row in self._sharing_cache:
# skip fieldnames if index == 0:
pass # skip fieldnames
if row['ShareType'] != ShareType: pass
pass if row['ShareType'] != ShareType:
elif row['PathOrToken'] != PathOrToken: pass
pass elif row['PathOrToken'] != PathOrToken:
else: pass
found = True else:
break found = True
index += 1 break
index += 1
if found: if found:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index) logger.debug("TRACE/sharing/%s/update: orig row[%d]=%r", ShareType, index, row)
# CSV: remove+adjust+readd # CSV: remove+adjust+readd
if PathMapped is not None: if PathMapped is not None:
row["PathMapped"] = PathMapped self._sharing_cache[index]["PathMapped"] = PathMapped
if Permissions is not None: if Permissions is not None:
row["Permissions"] = Permissions self._sharing_cache[index]["Permissions"] = Permissions
if User is not None: if User is not None:
row["User"] = User self._sharing_cache[index]["User"] = User
if EnabledByOwner is not None: if EnabledByOwner is not None:
row["EnabledByOwner"] = EnabledByOwner self._sharing_cache[index]["EnabledByOwner"] = EnabledByOwner
if EnabledByUser is not None: if EnabledByUser is not None:
row["EnabledByUser"] = EnabledByUser self._sharing_cache[index]["EnabledByUser"] = EnabledByUser
if HiddenByOwner is not None: if HiddenByOwner is not None:
row["HiddenByOwner"] = HiddenByOwner self._sharing_cache[index]["HiddenByOwner"] = HiddenByOwner
if HiddenByUser is not None: if HiddenByUser is not None:
row["HiddenByUser"] = HiddenByUser self._sharing_cache[index]["HiddenByUser"] = HiddenByUser
if Properties is not None: if Properties is not None:
row["Properties"] = Properties self._sharing_cache[index]["Properties"] = Properties
# update timestamp # update timestamp
row["TimestampUpdated"] = Timestamp self._sharing_cache[index]["TimestampUpdated"] = Timestamp
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: adj row[%d]=%r", ShareType, index, row) logger.debug("TRACE/sharing/%s/update: adj row[%d]=%r", ShareType, index, self._sharing_cache[index])
# replace row
self._sharing_cache[index] = row
with self._storage.acquire_lock("w", OwnerOrUser, path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file): if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: write CSV done", ShareType) logger.debug("TRACE/sharing/%s/update: write CSV done", ShareType)
return {"status": "success"} return {"status": "success"}
logger.error("sharing/%s/update: cannot update CSV database", ShareType)
return {"status": "error"} logger.error("sharing/%s/update: cannot update CSV database", ShareType)
else: return {"status": "error"}
return {"status": "not-found"} else:
return {"status": "not-found"}
def database_delete_sharing(self, def database_delete_sharing(self,
ShareType: str, ShareType: str,
@@ -344,42 +344,43 @@ class Sharing(sharing.BaseSharing):
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r", ShareType, PathOrToken) logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r", ShareType, PathOrToken)
# lookup token with self._storage.acquire_lock("w", path=self._sharing_db_file):
found = False # lookup token
index = 0 found = False
for row in self._sharing_cache: index = 0
if logger.isEnabledFor(logging.DEBUG): for row in self._sharing_cache:
logger.debug("TRACE/sharing/%s/delete: check: %r", ShareType, row) if logger.isEnabledFor(logging.DEBUG):
if index == 0: logger.debug("TRACE/sharing/%s/delete: check: %r", ShareType, row)
# skip fieldnames if index == 0:
pass # skip fieldnames
if row['ShareType'] != ShareType: pass
pass if row['ShareType'] != ShareType:
elif row['PathOrToken'] != PathOrToken: pass
pass elif row['PathOrToken'] != PathOrToken:
else: pass
found = True else:
break found = True
index += 1 break
index += 1
if found: if found:
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: found index=%d", ShareType, index) logger.debug("TRACE/sharing/%s/delete: found index=%d", ShareType, index)
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r index=%d", ShareType, PathOrToken, row['Owner'], index) logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r index=%d", ShareType, PathOrToken, row['Owner'], index)
self._sharing_cache.pop(index) self._sharing_cache.pop(index)
with self._storage.acquire_lock("w", row['Owner'], path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file): if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing_by_token: write CSV done") logger.debug("TRACE/sharing_by_token: write CSV done")
return {"status": "success"} return {"status": "success"}
logger.error("sharing/%s/delete: cannot update CSV database", ShareType)
return {"status": "error"}
else:
return {"status": "not-found"}
# local functions logger.error("sharing/%s/delete: cannot update CSV database", ShareType)
return {"status": "error"}
else:
return {"status": "not-found"}
# *** local functions ***
def _create_empty_csv(self, file: str) -> bool: def _create_empty_csv(self, file: str) -> bool:
with self._storage.acquire_lock("w", None, path=file): with self._storage.acquire_lock("w", None, path=file):
with open(file, 'w', newline='') as csvfile: with open(file, 'w', newline='') as csvfile:

2
radicale/sharing/files.py
View File

@@ -155,7 +155,7 @@ class Sharing(sharing.BaseSharing):
result = [] result = []
if logger.isEnabledFor(logging.DEBUG): 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) 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)
for _ShareType in sharing.SHARE_TYPES_V1: for _ShareType in sharing.SHARE_TYPES_V1:
if ShareType is not None and _ShareType != ShareType: if ShareType is not None and _ShareType != ShareType:

6
radicale/tests/test_base.py
View File

@@ -72,7 +72,11 @@ user: .*
collection: .* collection: .*
permissions: RrWw""") permissions: RrWw""")
self.configure({"rights": {"file": rights_file_path, self.configure({"rights": {"file": rights_file_path,
"type": "from_file"}}) "type": "from_file"},
"logging": {"request_header_on_debug": "True",
"request_content_on_debug": "True",
"response_header_on_debug": "True",
"response_content_on_debug": "True"}})
def test_root(self) -> None: def test_root(self) -> None:
"""GET request at "/".""" """GET request at "/"."""

287
radicale/tests/test_sharing.py
View File

@@ -191,6 +191,7 @@ class TestSharingApiSanity(BaseTest):
"collection_by_token": "True"}, "collection_by_token": "True"},
"rights": {"type": "owner_only"}}) "rights": {"type": "owner_only"}})
form_array: Sequence[str]
json_dict: dict json_dict: dict
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)): for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
@@ -231,7 +232,16 @@ class TestSharingApiSanity(BaseTest):
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
# check info hook # check info hook
logging.info("\n*** check API hook: info/all") logging.info("\n*** check API hook: info/all (text)")
form_array = []
_, headers, answer = self._sharing_api_form("all", "info", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status='success'" in answer
assert "PermittedCreateCollectionByMap=False" in answer
assert "PermittedCreateCollectionByToken=False" in answer
assert "FeatureEnabledCollectionByMap=True" in answer
assert "FeatureEnabledCollectionByToken=False" in answer
logging.info("\n*** check API hook: info/all (json)")
json_dict = {} json_dict = {}
_, headers, answer = self._sharing_api_json("all", "info", check=200, login="owner:ownerpw", json_dict=json_dict) _, headers, answer = self._sharing_api_json("all", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer) answer_dict = json.loads(answer)
@@ -306,6 +316,9 @@ class TestSharingApiSanity(BaseTest):
form_array: Sequence[str] form_array: Sequence[str]
json_dict: dict json_dict: dict
self.mkcalendar("/owner/collectionL1/", login="owner:ownerpw")
self.mkcalendar("/owner/collectionL2/", login="owner:ownerpw")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)): for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type) logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}}) self.configure({"sharing": {"type": db_type}})
@@ -319,13 +332,13 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** list (form->csv)") logging.info("\n*** list (form->csv)")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form(sharing_type, "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form(sharing_type, "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=not-found" in answer assert "Status='not-found'" in answer
assert "Lines=0" in answer assert "Lines=0" in answer
logging.info("\n*** list (json->text)") logging.info("\n*** list (json->text)")
json_dict = {} json_dict = {}
_, headers, answer = self._sharing_api_json(sharing_type, "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain") _, headers, answer = self._sharing_api_json(sharing_type, "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain")
assert "Status=not-found" in answer assert "Status='not-found'" in answer
assert "Lines=0" in answer assert "Lines=0" in answer
logging.info("\n*** list (json->json)") logging.info("\n*** list (json->json)")
@@ -338,10 +351,10 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** create a token -> 200") logging.info("\n*** create a token -> 200")
form_array = ["PathMapped=/owner/collectionL1/"] form_array = ["PathMapped=/owner/collectionL1/"]
_, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "PathOrToken=" in answer assert "PathOrToken='" in answer
# extract token # extract token
match = re.search('PathOrToken=(.+)', answer) match = re.search("PathOrToken='(.+)'", answer)
if match: if match:
token = match.group(1) token = match.group(1)
logging.info("received token %r", token) logging.info("received token %r", token)
@@ -359,21 +372,33 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** list/all (form->csv)") logging.info("\n*** list/all (form->csv)")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form("all", "list", check=200, login="owner:ownerpw", form_array=form_array, accept="text/csv")
assert "Status=" not in answer
assert "Line=" not in answer
assert "ShareType" in answer
assert "token" in answer
assert "map" in answer
logging.info("\n*** list/all (form->text)")
form_array = []
_, headers, answer = self._sharing_api_form("all", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("all", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=2" in answer assert "Lines=2" in answer
assert "Fields=" in answer
assert "Content[0]=" in answer
assert "Content[1]=" in answer
logging.info("\n*** delete token -> 200") logging.info("\n*** delete token -> 200")
form_array = ["PathOrToken=" + token] form_array = ["PathOrToken=" + token]
_, headers, answer = self._sharing_api_form("token", "delete", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "delete", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** delete share -> 200") logging.info("\n*** delete share -> 200")
form_array = [] form_array = []
form_array.append("PathOrToken=/user/collectionL2-shared-by-owner/") form_array.append("PathOrToken=/user/collectionL2-shared-by-owner/")
form_array.append("PathMapped=/owner/collectionL2/") form_array.append("PathMapped=/owner/collectionL2/")
_, headers, answer = self._sharing_api_form("map", "delete", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("map", "delete", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
def test_sharing_api_token_basic(self) -> None: def test_sharing_api_token_basic(self) -> None:
"""share-by-token API tests.""" """share-by-token API tests."""
@@ -394,6 +419,9 @@ class TestSharingApiSanity(BaseTest):
form_array: Sequence[str] form_array: Sequence[str]
json_dict: dict json_dict: dict
path_base1 = "/owner/collection1.ics/"
path_base2 = "/owner/collection2.ics/"
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)): for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type) logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}}) self.configure({"sharing": {"type": db_type}})
@@ -406,13 +434,21 @@ class TestSharingApiSanity(BaseTest):
json_dict = {} json_dict = {}
_, headers, answer = self._sharing_api_json("token", "create", 400, login="owner:ownerpw", json_dict=json_dict) _, headers, answer = self._sharing_api_json("token", "create", 400, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** create token#1 (form->text)") logging.info("\n*** create token#1 without existing collection (form->text)")
form_array = ["PathMapped=/owner/collection1/"] form_array = ["PathMapped=" + path_base1]
_, headers, answer = self._sharing_api_form("token", "create", check=404, login="owner:ownerpw", form_array=form_array)
logging.info("\n*** create collection*")
self.mkcalendar(path_base1, login="owner:ownerpw")
self.mkcalendar(path_base2, login="owner:ownerpw")
logging.info("\n*** create token#1 with existing collection (form->text)")
form_array = ["PathMapped=" + path_base1]
_, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "PathOrToken=" in answer assert "PathOrToken='" in answer
# extract token # extract token
match = re.search('PathOrToken=(.+)', answer) match = re.search("PathOrToken='(.+)'", answer)
if match: if match:
token1 = match.group(1) token1 = match.group(1)
logging.info("received token %r", token1) logging.info("received token %r", token1)
@@ -420,12 +456,12 @@ class TestSharingApiSanity(BaseTest):
assert False assert False
logging.info("\n*** create token#2 (json->text)") logging.info("\n*** create token#2 (json->text)")
json_dict = {'PathMapped': "/owner/collection2/"} json_dict = {'PathMapped': path_base2}
_, headers, answer = self._sharing_api_json("token", "create", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain") _, headers, answer = self._sharing_api_json("token", "create", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain")
assert "Status=success" in answer assert "Status='success'" in answer
assert "Token=" in answer assert "Token=" in answer
# extract token # extract token
match = re.search('Token=(.+)', answer) match = re.search("Token='(.+)'", answer)
if match: if match:
token2 = match.group(1) token2 = match.group(1)
logging.info("received token %r", token2) logging.info("received token %r", token2)
@@ -435,16 +471,16 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** lookup token#1 (form->text)") logging.info("\n*** lookup token#1 (form->text)")
form_array = ["PathOrToken=" + token1] form_array = ["PathOrToken=" + token1]
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
assert "/owner/collection1/" in answer assert path_base1 in answer
logging.info("\n*** lookup token#2 (json->text") logging.info("\n*** lookup token#2 (json->text")
json_dict = {'PathOrToken': token2} json_dict = {'PathOrToken': token2}
_, headers, answer = self._sharing_api_json("token", "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain") _, headers, answer = self._sharing_api_json("token", "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain")
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
assert "/owner/collection2/" in answer assert path_base2 in answer
logging.info("\n*** lookup token#2 (json->json)") logging.info("\n*** lookup token#2 (json->json)")
json_dict = {'PathOrToken': token2} json_dict = {'PathOrToken': token2}
@@ -452,46 +488,46 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer) answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success" assert answer_dict['Status'] == "success"
assert answer_dict['Lines'] == 1 assert answer_dict['Lines'] == 1
assert answer_dict['Content'][0]['PathMapped'] == "/owner/collection2/" assert answer_dict['Content'][0]['PathMapped'] == path_base2
logging.info("\n*** lookup tokens (form->text)") logging.info("\n*** lookup tokens (form->text)")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=2" in answer assert "Lines=2" in answer
assert "/owner/collection1/" in answer assert path_base1 in answer
assert "/owner/collection2/" in answer assert path_base2 in answer
logging.info("\n*** lookup tokens (form->csv)") logging.info("\n*** lookup tokens (form->csv)")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array, accept="text/csv") _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array, accept="text/csv")
assert "Status=success" not in answer assert "Status='success'" not in answer
assert "Lines=2" not in answer assert "Lines=2" not in answer
assert ";".join(sharing.DB_FIELDS_V1) in answer assert ";".join(sharing.DB_FIELDS_V1) in answer
assert "/owner/collection1/" in answer assert path_base1 in answer
assert "/owner/collection2/" in answer assert path_base2 in answer
logging.info("\n*** delete token#1 (form->text)") logging.info("\n*** delete token#1 (form->text)")
form_array = ["PathOrToken=" + token1] form_array = ["PathOrToken=" + token1]
_, headers, answer = self._sharing_api_form("token", "delete", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "delete", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** lookup token#1 (form->text) -> should not be there anymore") logging.info("\n*** lookup token#1 (form->text) -> should not be there anymore")
form_array = ["PathOrToken=" + token1] form_array = ["PathOrToken=" + token1]
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=not-found" in answer assert "Status='not-found'" in answer
assert "Lines=0" in answer assert "Lines=0" in answer
logging.info("\n*** lookup tokens (form->text) -> still one should be there") logging.info("\n*** lookup tokens (form->text) -> still one should be there")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
logging.info("\n*** disable token#2 as owner (form->text)") logging.info("\n*** disable token#2 as owner (form->text)")
form_array = ["PathOrToken=" + token2] form_array = ["PathOrToken=" + token2]
_, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** lookup token#2 (json->json) -> check for not enabled") logging.info("\n*** lookup token#2 (json->json) -> check for not enabled")
json_dict = {'PathOrToken': token2} json_dict = {'PathOrToken': token2}
@@ -512,7 +548,7 @@ class TestSharingApiSanity(BaseTest):
form_array = [] form_array = []
form_array.append("PathOrToken=" + token2) form_array.append("PathOrToken=" + token2)
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
assert "True;True;True;True" in answer assert "True;True;True;True" in answer
@@ -520,13 +556,13 @@ class TestSharingApiSanity(BaseTest):
form_array = [] form_array = []
form_array.append("PathOrToken=" + token2) form_array.append("PathOrToken=" + token2)
_, headers, answer = self._sharing_api_form("token", "hide", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "hide", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** lookup token#2 (form->text) -> check for hidden") logging.info("\n*** lookup token#2 (form->text) -> check for hidden")
form_array = [] form_array = []
form_array.append("PathOrToken=" + token2) form_array.append("PathOrToken=" + token2)
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
assert "True;True;True;True" in answer assert "True;True;True;True" in answer
@@ -561,6 +597,10 @@ class TestSharingApiSanity(BaseTest):
assert answer_dict['Status'] == "not-found" assert answer_dict['Status'] == "not-found"
assert answer_dict['Lines'] == 0 assert answer_dict['Lines'] == 0
logging.info("\n*** delete collection*")
self.delete(path_base1, login="owner:ownerpw")
self.delete(path_base2, login="owner:ownerpw")
def test_sharing_api_token_usage(self) -> None: def test_sharing_api_token_usage(self) -> None:
"""share-by-token API tests - real usage.""" """share-by-token API tests - real usage."""
self.configure({"auth": {"type": "htpasswd", self.configure({"auth": {"type": "htpasswd",
@@ -590,6 +630,8 @@ class TestSharingApiSanity(BaseTest):
path = path_base + "/event1.ics" path = path_base + "/event1.ics"
self.put(path, event, login="owner:ownerpw") self.put(path, event, login="owner:ownerpw")
self.mkcalendar(path_base2, login="owner:ownerpw")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)): for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type) logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}}) self.configure({"sharing": {"type": db_type}})
@@ -606,10 +648,10 @@ class TestSharingApiSanity(BaseTest):
form_array = [] form_array = []
form_array.append("PathMapped=" + path_base) form_array.append("PathMapped=" + path_base)
_, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "PathOrToken=" in answer assert "PathOrToken=" in answer
# extract token # extract token
match = re.search('PathOrToken=(.+)', answer) match = re.search("PathOrToken='(.+)'", answer)
if match: if match:
token = match.group(1) token = match.group(1)
logging.info("received token %r", token) logging.info("received token %r", token)
@@ -620,10 +662,10 @@ class TestSharingApiSanity(BaseTest):
form_array = [] form_array = []
form_array.append("PathMapped=" + path_base2) form_array.append("PathMapped=" + path_base2)
_, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "PathOrToken=" in answer assert "PathOrToken=" in answer
# extract token # extract token
match = re.search('PathOrToken=(.+)', answer) match = re.search("PathOrToken='(.+)'", answer)
if match: if match:
token2 = match.group(1) token2 = match.group(1)
logging.info("received token %r", token2) logging.info("received token %r", token2)
@@ -633,7 +675,7 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** enable token (form->text)") logging.info("\n*** enable token (form->text)")
form_array = ["PathOrToken=" + token] form_array = ["PathOrToken=" + token]
_, headers, answer = self._sharing_api_form("token", "enable", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "enable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** fetch collection using invalid token (without credentials)") 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", path_token + "v1/invalidtoken", check=401)
@@ -645,7 +687,7 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** disable token (form->text)") logging.info("\n*** disable token (form->text)")
form_array = ["PathOrToken=" + token] form_array = ["PathOrToken=" + token]
_, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** fetch collection using disabled token (without credentials)") 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", path_token + token, check=401)
@@ -653,7 +695,7 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** enable token (form->text)") logging.info("\n*** enable token (form->text)")
form_array = ["PathOrToken=" + token] form_array = ["PathOrToken=" + token]
_, headers, answer = self._sharing_api_form("token", "enable", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("token", "enable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
logging.info("\n*** fetch collection using token (without credentials)") logging.info("\n*** fetch collection using token (without credentials)")
_, headers, answer = self.request("GET", path_token + token, check=200) _, headers, answer = self.request("GET", path_token + token, check=200)
@@ -2207,7 +2249,7 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** list/all (form->csv)") logging.info("\n*** list/all (form->csv)")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form("map", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("map", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
# read collection # read collection
@@ -2227,7 +2269,7 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** list/all (form->csv)") logging.info("\n*** list/all (form->csv)")
form_array = [] form_array = []
_, headers, answer = self._sharing_api_form("map", "list", check=200, login="owner:ownerpw", form_array=form_array) _, headers, answer = self._sharing_api_form("map", "list", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
assert "Lines=1" in answer assert "Lines=1" in answer
# read collection # read collection
@@ -2415,8 +2457,23 @@ class TestSharingApiSanity(BaseTest):
"collection_by_token": "True"}, "collection_by_token": "True"},
"logging": {"request_header_on_debug": "False", "logging": {"request_header_on_debug": "False",
"response_content_on_debug": "False", "response_content_on_debug": "False",
"rights_rule_doesnt_match_on_debug": "True",
"request_content_on_debug": "True"}, "request_content_on_debug": "True"},
"rights": {"type": "owner_only"}}) "rights": {"type": "owner_only"}})
rights_file_path = os.path.join(self.colpath, "rights")
with open(rights_file_path, "w") as f:
f.write("""\
[default-collection]
user: .+
collection: .+
permissions: RrWw
[default]
user: .+
collection: {user}(/.*)?
permissions: RrWw""")
self.configure({"rights": {"file": rights_file_path}})
json_dict: dict json_dict: dict
path_user1 = "/user1/calendarCCu1.ics/" path_user1 = "/user1/calendarCCu1.ics/"
@@ -2440,6 +2497,9 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** test: %s", db_type) logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}}) self.configure({"sharing": {"type": db_type}})
# owner_only
self.configure({"rights": {"type": "owner_only"}})
# create map # create map
logging.info("\n*** create map user1/owner1 -> ok") logging.info("\n*** create map user1/owner1 -> ok")
json_dict = {} json_dict = {}
@@ -2453,7 +2513,7 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer) answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success" assert answer_dict['Status'] == "success"
logging.info("\n*** mkcalendar user1 for shared -> conflict") logging.info("\n*** mkcalendar as user1 for user1/shared1 -> conflict")
self.mkcalendar(path_user1_shared1, login="user1:user1pw", check=409) self.mkcalendar(path_user1_shared1, login="user1:user1pw", check=409)
# create map # create map
@@ -2469,7 +2529,7 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer) answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success" assert answer_dict['Status'] == "success"
logging.info("\n*** mkcol user2 for shared -> conflict") logging.info("\n*** mkcol as user2 for user2/shared1 -> conflict")
self.mkcalendar(path_user2_shared1, login="user2:user2pw", check=409) self.mkcalendar(path_user2_shared1, login="user2:user2pw", check=409)
# create map # create map
@@ -2483,6 +2543,15 @@ class TestSharingApiSanity(BaseTest):
json_dict['Hidden'] = False json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=409, login="owner1:owner1pw", json_dict=json_dict) _, headers, answer = self._sharing_api_json("map", "create", check=409, login="owner1:owner1pw", json_dict=json_dict)
# from_file
self.configure({"rights": {"type": "from_file"}})
logging.info("\n*** mkcalendar as user1 for user2/shared1 with rights from file -> conflict")
self.mkcalendar(path_user2_shared1, login="user1:user1pw", check=409)
logging.info("\n*** mkcol as user1 for user2/shared1 with rights from file -> conflict")
self.mkcol(path_user2_shared1, login="user1:user1pw", check=409)
def test_sharing_api_permissions_global(self) -> None: def test_sharing_api_permissions_global(self) -> None:
"""sharing API usage tests related to global permissions.""" """sharing API usage tests related to global permissions."""
self.configure({"auth": {"type": "htpasswd", self.configure({"auth": {"type": "htpasswd",
@@ -2916,7 +2985,7 @@ permissions: RrWw""")
json_dict['PathOrToken'] = path_shared_r json_dict['PathOrToken'] = path_shared_r
_, headers, answer = self._sharing_api_json("map", "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)
# verify PROPPATCH as user # verify PROPFIND as user
logging.info("\n*** PROPFIND collection user -> ok") logging.info("\n*** PROPFIND collection user -> ok")
propfind_calendar_color = get_file_content("propfind_multiple.xml") propfind_calendar_color = get_file_content("propfind_multiple.xml")
_, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw") _, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw")
@@ -2987,7 +3056,7 @@ permissions: RrWw""")
form_array.append("Properties='C:calendar-description'='ICAL-USER-NEW'") form_array.append("Properties='C:calendar-description'='ICAL-USER-NEW'")
form_array.append("Properties='ICAL:calendar-color'='#CCCCCC'") form_array.append("Properties='ICAL:calendar-color'='#CCCCCC'")
_, headers, answer = self._sharing_api_form("map", "update", check=200, login="user:userpw", form_array=form_array) _, headers, answer = self._sharing_api_form("map", "update", check=200, login="user:userpw", form_array=form_array)
assert "Status=success" in answer assert "Status='success'" in answer
# verify overlay as user # verify overlay as user
logging.info("\n*** PROPFIND collection user (overlay) -> ok") logging.info("\n*** PROPFIND collection user (overlay) -> ok")
@@ -3004,12 +3073,132 @@ permissions: RrWw""")
assert status == 200 and prop.text == "#CCCCCC" assert status == 200 and prop.text == "#CCCCCC"
# update map by user # update map by user
logging.info("\n*** update map by user (form)") logging.info("\n*** update properties with buggyy ones by user (form)")
form_array = ["User=" + "user"] form_array = ["User=" + "user"]
form_array.append("PathOrToken=" + path_shared_r) form_array.append("PathOrToken=" + path_shared_r)
form_array.append("Properties=BUGGYENTRY=BUGGYVALUE") form_array.append("Properties=BUGGYENTRY=BUGGYVALUE")
_, headers, answer = self._sharing_api_form("map", "update", check=400, login="user:userpw", form_array=form_array) _, headers, answer = self._sharing_api_form("map", "update", check=400, login="user:userpw", form_array=form_array)
def test_sharing_api_map_propfind_overlay_api_delete(self) -> None:
"""share-by-map API usage tests related to proppatch."""
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,
"permit_properties_overlay": True,
"collection_by_map": "True",
"collection_by_token": "True"},
"logging": {"request_header_on_debug": "False",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"rights": {"type": "owner_only"}})
form_array: Sequence[str]
json_dict: dict
path_mapped = "/owner/calendarPFD.ics/"
path_shared_r = "/user/calendarPFD-shared-by-owner-r.ics/"
logging.info("\n*** prepare and test access")
self.mkcalendar(path_mapped, 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}})
# 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:">
<prop>
<current-user-principal />
</prop>
</propfind>""", login="owner:ownerpw")
logging.info("response: %r", responses)
response = responses[path_mapped]
assert not isinstance(response, int) and len(response) == 1
status, prop = response["D:current-user-principal"]
assert status == 200 and len(prop) == 1
element = prop.find(xmlutils.make_clark("D:href"))
assert element is not None and element.text == "/owner/"
# create map
logging.info("\n*** create map user/owner:r -> ok")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
# enable map by user
logging.info("\n*** enable map 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("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
# set properties by user
logging.info("\n*** set properties by user (form)")
form_array = []
form_array.append("PathOrToken=" + path_shared_r)
form_array.append("Properties='ICAL:calendar-color'='#CCCCCC'")
_, headers, answer = self._sharing_api_form("map", "update", check=200, login="user:userpw", form_array=form_array)
# check that properties are existing in map
logging.info("\n*** list and check for properties (json->json)")
_, 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
assert answer_dict['Content'][0]['Properties']['ICAL:calendar-color'] == '#CCCCCC'
# verify overlay as user
logging.info("\n*** PROPFIND collection user (overlay) -> ok")
propfind_calendar_color = get_file_content("propfind_calendar_color.xml")
_, responses = self.propfind(path_shared_r, propfind_calendar_color, login="user:userpw")
logging.info("response: %r", responses)
response = responses[path_shared_r]
assert not isinstance(response, int)
status, prop = response["ICAL:calendar-color"]
logging.debug("calendar-color: %r", prop.text)
assert status == 200 and prop.text == "#CCCCCC"
# clear properties by user
logging.info("\n*** clear properties by user (form)")
form_array = []
form_array.append("PathOrToken=" + path_shared_r)
form_array.append("Properties=")
_, headers, answer = self._sharing_api_form("map", "update", check=200, login="user:userpw", form_array=form_array)
# check that properties are cleared
logging.info("\n*** list and check for cleared properties (json->json)")
_, 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
assert answer_dict['Content'][0]['Properties'] is not None
# verify overlay as user
logging.info("\n*** PROPFIND collection user (overlay no longer exists) -> ok")
propfind_calendar_color = get_file_content("propfind_calendar_color.xml")
_, responses = self.propfind(path_shared_r, propfind_calendar_color, login="user:userpw")
logging.info("response: %r", responses)
response = responses[path_shared_r]
assert not isinstance(response, int)
status, prop = response["ICAL:calendar-color"]
logging.debug("calendar-color: %r", prop.text)
assert status == 404
def test_sharing_api_map_propfind_overlay_api_permissions(self) -> None: def test_sharing_api_map_propfind_overlay_api_permissions(self) -> None:
"""share-by-map API usage tests related to proppatch.""" """share-by-map API usage tests related to proppatch."""
self.configure({"auth": {"type": "htpasswd", self.configure({"auth": {"type": "htpasswd",