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

Merge pull request #2011 from pbiering/sharing-review-1

Browse Source 
Sharing review #1
This commit is contained in:
Peter Bieringer
2026-03-03 20:57:32 +01:00
committed by GitHub
parent f9399dc17e 3bb333a1a7
commit 460b249cdc
6 changed files with 832 additions and 688 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
SHARING.md
View File

@@ -167,18 +167,31 @@ curl -u user:pass --silent -H "accept: application/json" -d "" http://localhost:
```
##### API Hook "*/create"
##### API Hook "(token|map)/create"
Create a share
* Authorization
Authenticated user is `Owner`
###### API Hook "token/create"
Create a share by mapping a collection of an `Owner` to a token.
* Authorization
Authenticated user as `Owner` has at least read access to `PathMapped`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathMapped | yes | |
| Permissions | no | r
| Parameter | Owner |
| - | - |
| Owner | implicit(by authentication) |
| PathMapped | mandatory |
| User | optional(default:owner) |
| Permissions | optional(default:r) |
| Enabled | optional(owner) |
| Hidden | optional(owner) |
| Properties | optional |
* Output
@@ -195,17 +208,28 @@ Status=success
PathOrToken=v1/VQR7AmsVRi2ZlFj_JwGpFx-ES5Goyku-gP_YkLh1zUw=
```
###### API Hook "map/create"
Create a share by mapping a collection of an `Owner` to an `User`.
* Authorization
Authenticated user as `Owner` has at least read access to `PathMapped`
Provided `User` has at least read access to `PathOrToken`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | |
| PathMapped | yes | |
| Permissions | no | r
| User | yes | |
| Parameter | Value |
| - | - |
| Owner | implicit(by authentication) |
| PathOrToken | mandatory |
| PathMapped | mandatory |
| User | mandatory |
| Permissions | optional(default:r) |
| Enabled | optional(owner) |
| Hidden | optional(owner) |
| Properties | optional |
* Output: result status
@@ -218,19 +242,24 @@ Status=success
```
##### API Hook "*/list"
##### API Hook "(map|token|all)/list"
List shares (optional with filter)
List shares (optional with filter) either owned or assigned as user.
* Authorization
Authenticated user as `Owner` or `User`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathMapped | no | (all) |
| Owner | no | (owned ones) |
| User | no | (filtered) |
| Parameter | Filter |
| - | - |
| Owner | implicit(by authentication) |
| User | implicit(by authentication) |
| PathOrToken | optional |
| PathMapped | optional |
* Output: plain/csv/json
* Output: plain/csv/json
* Example: CSV
@@ -241,45 +270,62 @@ map,/user/cal1-from-owner/,/owner/cal1/,owner,user,r,False,False,True,True,17719
```
##### API Hook "*/delete"
##### API Hook "(map|token)/delete"
Delete a share selected by `PathOrToken`
Delete a share selected by `PathOrToken`.
* Authorization
Authenticated user is `Owner`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | |
| Parameter | Type | Owner | User |
| - | - | - | - |
| PathOrToken | selector | mandatory | not-permitted |
* Output: result status
##### API Hook "*/update"
##### API Hook "(token|map)/update"
Update a share selected by `PathOrToken`
Update a share selected by `PathOrToken`.
Execute delete+create in case `PathOrToken` needs to be changed.
* Authorization
Authenticated user is `Owner` or `User`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | n/a |
| PathMapped | no | |
| OwnerOrUser | yes | n/a |
| User | no | |
| Properties | no | |
| Parameter | Type | Owner | User |
| - | - | - | - |
| PathOrToken | selector | mandatory | mandatory |
| Owner | by authentication | not-permitted | not-permitted |
| PathMapped | adjustable | optional | not-permitted |
| User | adjustable | optional | not-permitted |
| Permissions | adjustable | optional | not-permitted |
| Enabled | adjustable | optional(owner) | optional(user) |
| Hidden | adjustable | optional(owner) | optional(user) |
| Properties | adjustable | optional | optional |
* Output: result status
##### API Hooks "*/(enable|disable|hide|unhide)"
##### API Hooks "(map|token)/(enable|disable|hide|unhide)"
Enable|disable|hide|unhide a share selected by `PathOrToken`
Toggle enable|disable|hide|unhide of `Owner` or `User` of a share selected by `PathOrToken`
* Authorization
Authenticated user is `Owner` or `User`
* Input
| Parameter | Mandatory | Default |
| - | - | - |
| PathOrToken | yes | n/a |
| Parameter | Type | Owner | User |
| - | - | - | - |
| PathOrToken | selector | mandatory | mandatory |
* Output: result status
* Output: result status
* Example: TEXT (enable)

8
radicale/app/proppatch.py
View File

@@ -80,10 +80,10 @@ def xml_proppatch(base_prefix: str, path: str,
all_props = radicale_item.check_and_sanitize_props(all_props_with_remove)
if sharing and sharing_overlay and _sharing is not None:
# _sharing is not None: always the case, but makes mypy happy
_sharing.update_sharing(ShareType=sharing['ShareType'],
PathOrToken=sharing['PathOrToken'],
OwnerOrUser=sharing['User'],
Properties=cast(Dict[str, str], all_props))
_sharing.database_update_sharing(ShareType=sharing['ShareType'],
PathOrToken=sharing['PathOrToken'],
OwnerOrUser=sharing['User'],
Properties=cast(Dict[str, str], all_props))
else:
if collection is not None:
# always the case, but makes mypy happy

498
radicale/sharing/__init__.py
View File

@@ -36,6 +36,7 @@ INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none")
DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated', 'Properties')
DB_FIELDS_V1_BOOL: Sequence[str] = ('EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser')
DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
DB_FIELDS_V1_USER_PERMITTED: Sequence[str] = ('EnabledByUser', 'HiddenByUser', 'Properties')
# ShareType: <token|map>
# PathOrToken: <path|token> [PrimaryKey]
# PathMapped: <path>
@@ -48,6 +49,7 @@ DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
# HiddenByUser: True|False (share exposure controlled by user) - check skipped if Owner==User
# TimestampCreated: <unixtime> (when created)
# TimestampUpdated: <unixtime> (last update)
# Properties: Overlay of collection properties
SHARE_TYPES: Sequence[str] = ('token', 'map', 'all')
SHARE_TYPES_V1: Sequence[str] = ('token', 'map')
@@ -89,6 +91,8 @@ class BaseSharing:
_storage: storage.BaseStorage
_rights: rights.BaseRights
_enabled: bool = False
default_permissions_create_token: str
default_permissions_create_map: str
def __init__(self, configuration: "config.Configuration") -> None:
"""Initialize Sharing.
@@ -128,106 +132,96 @@ class BaseSharing:
# database tasks
self.sharing_db_type = configuration.get("sharing", "type")
logger.info("sharing.db_type: %s", self.sharing_db_type)
logger.info("sharing.database_type: %s", self.sharing_db_type)
try:
if self.init_database() is False:
if self.database_init() is False:
logger.info("sharing disabled as no database is active")
self._enabled = False
return
except Exception as e:
logger.error("sharing database cannot be initialized: %r", e)
exit(1)
database_info = self.get_database_info()
database_info = self.database_get_info()
if database_info:
logger.info("sharing database info: %r", database_info)
else:
logger.info("sharing database info: (not provided)")
# overloadable functions
def init_database(self) -> bool:
""" initialize database """
# overloadable database functions
def database_init(self) -> bool:
""" initialize db """
return False
def get_database_info(self) -> Union[dict, None]:
""" retrieve database information """
def database_get_info(self) -> Union[dict, None]:
""" retrieve db information """
return None
def verify_database(self) -> bool:
""" verify database information """
def database_verify(self) -> bool:
""" verify db information """
return False
def list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
def database_list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
""" retrieve sharing """
return []
def get_sharing(self,
ShareType: str,
PathOrToken: str,
User: Union[str, None] = None) -> Union[dict, None]:
def database_get_sharing(self,
ShareType: str,
PathOrToken: str,
OnlyEnabled: bool = True,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
return {"status": "not-implemented"}
def create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
def database_create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
""" create sharing """
return {"status": "not-implemented"}
def update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
def database_update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: Union[str, None] = None,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
""" update sharing """
return {"status": "not-implemented"}
def delete_sharing(self,
ShareType: str,
PathOrToken: str,
Owner: str,
PathMapped: Union[str, None] = None) -> dict:
def database_delete_sharing(self,
ShareType: str,
PathOrToken: str) -> dict:
""" delete sharing """
return {"status": "not-implemented"}
def toggle_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
Action: str,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
Timestamp: int = 0) -> dict:
""" toggle sharing """
return {"status": "not-implemented"}
# sharing functions called by request methods
def verify(self) -> bool:
""" verify database """
logger.info("sharing database verification begin")
logger.info("sharing database verification call: %s", self.sharing_db_type)
result = self.verify_database()
result = self.database_verify()
if result is not True:
logger.error("sharing database verification call -> PROBLEM: %s", self.sharing_db_type)
return False
@@ -237,7 +231,7 @@ class BaseSharing:
# check all entries
logger.info("sharing database verification content start")
with self._storage.acquire_lock("r"):
for entry in self.list_sharing():
for entry in self.database_list_sharing():
logger.debug("analyze: %r", entry)
if entry['ShareType'] not in SHARE_TYPES_V1:
logger.error("ShareType not supported: %r", entry['ShareType'])
@@ -291,7 +285,7 @@ class BaseSharing:
# retrieve collections which are enabled and not hidden by owner+user
if active:
shared_collection_list = self.list_sharing(
shared_collection_list = self.database_list_sharing(
ShareType="map",
OwnerOrUser=user,
User=user,
@@ -301,7 +295,7 @@ class BaseSharing:
HiddenByUser=False)
else:
# unconditional
shared_collection_list = self.list_sharing(
shared_collection_list = self.database_list_sharing(
ShareType="map",
OwnerOrUser=user,
User=user)
@@ -326,7 +320,7 @@ class BaseSharing:
# TODO add token validity checks
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: supported token found in path: %r (token=%r)", path, match[1])
return self.get_sharing(
return self.database_get_sharing(
ShareType="token",
PathOrToken=match[1])
else:
@@ -343,7 +337,7 @@ class BaseSharing:
if self.sharing_collection_by_map:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/resolver: check path: %r", path)
result = self.get_sharing(
result = self.database_get_sharing(
ShareType="map",
PathOrToken=path,
User=user)
@@ -354,7 +348,7 @@ class BaseSharing:
parent_path = pathutils.parent_path(path)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map/resolver: check parent path: %r", parent_path)
result = self.get_sharing(
result = self.database_get_sharing(
ShareType="map",
PathOrToken=parent_path,
User=user)
@@ -484,6 +478,12 @@ class BaseSharing:
request_data = json.loads(request_body)
except json.JSONDecodeError:
return httputils.bad_request("Invalid JSON")
for key in ["Enabled", "Hidden"]:
# convert JSON boolean
if key in request_data:
if type(request_data[key]) is not bool:
logger.error(api_info + ": unsupported (non-boolean) " + key + ": " + request_data[key])
return httputils.bad_request("Invalid non-boolean value for " + key + ": " + request_data[key])
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + " (json): %r", f"{request_data}")
elif 'application/x-www-form-urlencoded' in content_type:
@@ -504,6 +504,12 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: converted Properties from form into dict: %r", properties_dict)
request_data[key] = properties_dict
elif key in ["Enabled", "Hidden"]:
try:
request_data[key] = config._convert_to_bool(request_parsed[key][0])
except ValueError:
logger.error(api_info + ": unsupported (non-boolean) " + key + ": " + request_parsed[key][0])
return httputils.bad_request("Invalid non-boolean value for " + key + ": " + request_parsed[key][0])
else:
request_data[key] = request_parsed[key][0]
if logger.isEnabledFor(logging.DEBUG):
@@ -538,17 +544,16 @@ class BaseSharing:
Owner: str = user
User: Union[str, None] = None
Permissions: Union[str, None] = None # no permissions by default
EnabledByOwner: Union[bool, None] = None
HiddenByOwner: Union[bool, None] = None
EnabledByUser: Union[bool, None] = None
HiddenByUser: Union[bool, None] = None
Enabled: Union[bool, None] = None
Hidden: Union[bool, None] = None
Properties: Union[dict, None] = None
# parameters sanity check
for key in request_data:
if key == "Permissions":
if not re.search('^[a-zA-Z]+$', request_data[key]):
return httputils.bad_request("Invalid value for Permissions")
for permission in request_data[key]:
if permission not in rights.INTERNAL_PERMISSIONS:
return httputils.bad_request("Invalid value for Permissions")
elif key == "PathOrToken":
if ShareType == "token":
if not re.search('^' + TOKEN_PATTERN_V1 + '$', request_data[key]):
@@ -566,28 +571,14 @@ class BaseSharing:
return httputils.bad_request("Invalid value for PathMapped")
elif not request_data[key].endswith("/"):
return httputils.bad_request("PathMapped not ending with /")
elif key == "Enabled" or key == "Hidden":
if not re.search('^(False|True)$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for " + key)
elif key == "User":
if not re.search('^' + USER_PATTERN + '$', request_data[key]):
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for User")
# check for mandatory parameters
if 'PathMapped' not in request_data:
if action in ['info', 'list', 'update']:
# ignored
pass
else:
if ShareType == "token" and action != 'create':
# optional
pass
else:
logger.error(api_info + ": missing PathMapped")
return httputils.bad_request("Missing PathMapped")
else:
# check for optional parameters
if 'PathMapped' in request_data:
# used by create or list(filter)
PathMapped = request_data['PathMapped']
if 'PathOrToken' not in request_data:
@@ -617,20 +608,20 @@ class BaseSharing:
return httputils.bad_request("Property not supported to overlay: %r" % entry)
Properties = request_data['Properties']
if ShareType == "map":
if action == 'info':
# ignored
pass
else:
if 'User' not in request_data:
if action not in ['list', 'delete', 'update']:
logger.warning(api_info + ": missing User")
return httputils.bad_request("Missing User")
else:
# optional
pass
else:
User = request_data['User']
if 'Enabled' in request_data:
Enabled = request_data['Enabled']
else:
Enabled = None
if 'Hidden' in request_data:
Hidden = request_data['Hidden']
else:
Hidden = None
if 'User' in request_data:
User = request_data['User']
else:
User = None
answer: dict = {}
result: dict = {}
@@ -647,20 +638,20 @@ class BaseSharing:
if action == "list":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if 'PathOrToken' in request_data:
PathOrToken = request_data['PathOrToken']
if PathOrToken is not None:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": filter: %r", PathOrToken)
if ShareType != "all":
result_array = self.list_sharing(
result_array = self.database_list_sharing(
ShareType=ShareType,
OwnerOrUser=Owner,
OwnerOrUser=user,
PathMapped=PathMapped,
PathOrToken=PathOrToken)
else:
result_array = self.list_sharing(
OwnerOrUser=Owner,
result_array = self.database_list_sharing(
OwnerOrUser=user,
PathMapped=PathMapped,
PathOrToken=PathOrToken)
@@ -675,7 +666,12 @@ class BaseSharing:
elif action == "create":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if 'Permissions' not in request_data:
if PathMapped is None:
logger.error(api_info + ": missing PathMapped")
return httputils.bad_request("Missing PathMapped")
if Permissions is None:
if ShareType == "token":
Permissions = self.default_permissions_create_token
elif ShareType == "map":
@@ -683,23 +679,27 @@ class BaseSharing:
else:
# default
Permissions = "r"
if 'Enabled' in request_data:
EnabledByOwner = config._convert_to_bool(request_data['Enabled'])
else:
EnabledByOwner = False # security by default
Permissions = str(Permissions)
if 'Hidden' in request_data:
HiddenByOwner = config._convert_to_bool(request_data['Hidden'])
else:
HiddenByOwner = True # security by default
if Enabled is None:
Enabled = False # security by default
EnabledByUser = False # security by default
HiddenByUser = True # security by default
if Hidden is None:
Hidden = True # security by default
# create token share with security-by-default for User
EnabledByUser: bool = False
HiddenByUser: bool = True
if user == User:
# create token share with same flags
EnabledByUser = Enabled
HiddenByUser = Hidden
if ShareType == "token":
# check access Permissions
access = Access(self._rights, user, str(PathMapped)) # PathMapped is mandatory
access = Access(self._rights, user, PathMapped)
if not access.check("r"):
logger.info("Add sharing-by-token: access to %r not allowed for user %r", PathMapped, user)
return httputils.NOT_ALLOWED
@@ -713,20 +713,32 @@ class BaseSharing:
logger.info("Add sharing-by-token: access to %r not allowed for user %r (permit=True but denied by 'T')", PathMapped, user)
return httputils.NOT_ALLOWED
if User is not None:
# user is optional on tokens, otherwise it's the owner itself
User = str(User)
else:
User = user
# v1: create uuid token with 2x 32 bytes = 256 bit
token = "v1/" + str(base64.urlsafe_b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes), 'utf-8')
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": %r (Permissions=%r token=%r)", PathMapped, Permissions, token)
result = self.create_sharing(
result = self.database_create_sharing(
ShareType=ShareType,
PathOrToken=token,
PathMapped=str(PathMapped), # mandatory
Owner=Owner, User=Owner,
Permissions=str(Permissions), # mandantory
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
PathMapped=PathMapped,
Owner=user,
User=User,
Permissions=Permissions,
EnabledByOwner=Enabled,
EnabledByUser=EnabledByUser,
HiddenByOwner=Hidden,
HiddenByUser=HiddenByUser,
Timestamp=Timestamp,
Properties=Properties)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": result=%r", result)
@@ -737,13 +749,19 @@ class BaseSharing:
else:
PathOrToken = str(PathOrToken)
# retrieve existing share
share = self.database_get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False)
if share is not None:
logger.error("Sharing/create/%s: already exists: %r", ShareType, PathOrToken)
return httputils.CONFLICT
if User is None:
return httputils.bad_request("Missing User")
else:
User = str(User)
# check access Permissions
access = Access(self._rights, Owner, str(PathMapped), None) # PathMapped is mandatory
access = Access(self._rights, Owner, PathMapped, None) # PathMapped is mandatory
if not access.check("r") and "i" not in access.permissions:
logger.info("Add sharing-by-map: access to path(mapped) %r not allowed for owner %r", PathMapped, Owner)
return httputils.NOT_ALLOWED
@@ -757,7 +775,7 @@ class BaseSharing:
logger.info("Add sharing-by-map: access to %r not allowed for user %r (permit=True but denied by 'M')", PathMapped, user)
return httputils.NOT_ALLOWED
access = Access(self._rights, str(User), PathOrToken)
access = Access(self._rights, User, PathOrToken)
if not access.check("r"):
logger.info("Add sharing-by-map: access to path %r not allowed for user %r", PathOrToken, User)
return httputils.NOT_ALLOWED
@@ -773,15 +791,18 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": %r (Permissions=%r PathOrToken=%r user=%r)", PathMapped, Permissions, PathOrToken, User)
result = self.create_sharing(
result = self.database_create_sharing(
ShareType=ShareType,
PathOrToken=PathOrToken, # verification above that it is not None
PathMapped=str(PathMapped), # mandatory
Owner=Owner,
User=User, # verification above that it is not None
Permissions=str(Permissions), # mandatory
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
EnabledByUser=EnabledByUser, HiddenByUser=HiddenByUser,
PathOrToken=PathOrToken,
PathMapped=PathMapped,
Owner=user,
User=User,
Permissions=Permissions,
EnabledByOwner=Enabled,
EnabledByUser=EnabledByUser,
HiddenByOwner=Hidden,
HiddenByUser=HiddenByUser,
Timestamp=Timestamp,
Properties=Properties)
@@ -809,66 +830,74 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
if ShareType not in ["token", "map"]:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
# check for permissions to update
share = self.get_sharing(ShareType=ShareType, PathOrToken=PathOrToken)
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
else:
PathOrToken = str(PathOrToken)
# retrieve existing share
share = self.database_get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False)
if share is None:
return httputils.NOT_FOUND
if share['Owner'] is not None and user == share['Owner']:
# unconditional update as owner
result = self.update_sharing(
if user == share['Owner']:
if PathMapped is not None:
# check access Permissions
access = Access(self._rights, user, str(PathMapped), None)
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)
return httputils.NOT_ALLOWED
result = self.database_update_sharing(
ShareType=ShareType,
PathMapped=PathMapped,
Permissions=Permissions,
EnabledByOwner=EnabledByOwner,
HiddenByOwner=HiddenByOwner,
PathOrToken=str(PathOrToken), # verification above that it is not None
EnabledByOwner=Enabled,
HiddenByOwner=Hidden,
PathOrToken=PathOrToken,
OwnerOrUser=user,
User=User,
Timestamp=Timestamp,
Properties=Properties)
elif share['User'] is not None and Owner == share['User']:
elif user == share['User']:
# User is only allowed to update Properties
if PathMapped is not None or EnabledByOwner is not None or HiddenByOwner is not None:
logger.info("Update sharing: access to %r not allowed for user %r to adjust anything beside Properties", PathOrToken, user)
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))
return httputils.NOT_ALLOWED
if Properties is None:
logger.info("Update sharing: access to %r as user %r misses Properties", PathOrToken, user)
return httputils.NOT_ALLOWED
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/update: permit_properties_overlay=%s Permissions=%r", self.permit_properties_overlay, share['Permissions'])
if self.permit_properties_overlay:
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)
return httputils.NOT_ALLOWED
if Properties is not None:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/update: permit_properties_overlay=%s Permissions=%r", self.permit_properties_overlay, share['Permissions'])
if self.permit_properties_overlay:
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)
return httputils.NOT_ALLOWED
else:
logger.info("Update on shared %r: overlay permitted by option", PathOrToken)
else:
logger.info("Update on shared %r: overlay permitted by option", PathOrToken)
else:
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)
else:
logger.info("Update on shared %r: overlay denied by option", PathOrToken)
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)
else:
logger.info("Update on shared %r: overlay denied by option", PathOrToken)
return httputils.NOT_ALLOWED
return httputils.NOT_ALLOWED
return httputils.NOT_ALLOWED
# limited update as user
result = self.update_sharing(
result = self.database_update_sharing(
ShareType=ShareType,
PathMapped=PathMapped,
PathOrToken=str(PathOrToken), # verification above that it is not None
OwnerOrUser=user,
EnabledByUser=Enabled,
HiddenByUser=Hidden,
Timestamp=Timestamp,
Properties=Properties)
else:
# neither owner nor user matches
logger.warning("Update sharing of %r not permitted for user %r", PathOrToken, user)
return httputils.NOT_ALLOWED
# result handling
@@ -891,26 +920,29 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
if ShareType == "token":
result = self.delete_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
Owner=Owner)
elif ShareType == "map":
result = self.delete_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
PathMapped=PathMapped,
Owner=Owner)
else:
if ShareType not in ["token", "map"]:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
else:
PathOrToken = str(PathOrToken)
# check whether share exists
share = self.database_get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False)
if share is None:
return httputils.NOT_FOUND
if user == share['Owner']:
result = self.database_delete_sharing(
ShareType=ShareType,
PathOrToken=PathOrToken) # verification above that it is not None
else:
# only owner is permitted to delete a share
logger.warning("Delete sharing of %r not permitted for user %r", PathOrToken, user)
return httputils.NOT_ALLOWED
# result handling
if result['status'] == "not-found":
return httputils.NOT_FOUND
@@ -941,35 +973,75 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API/POST/" + action)
if ShareType in ["token", "map"]:
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
result = self.toggle_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
OwnerOrUser=user, # authenticated user
User=User, # optional for selection
PathMapped=PathMapped, # optional for selection
Action=action,
Timestamp=Timestamp)
if result:
if result['status'] == "not-found":
return httputils.NOT_FOUND
if result['status'] == "permission-denied":
return httputils.NOT_ALLOWED
elif result['status'] == "success":
answer['Status'] = "success"
pass
else:
logger.error("Toggle sharing: %r of user %s not successful", request_data['PathOrToken'], user)
return httputils.bad_request("Internal Error")
else:
if ShareType not in ["token", "map"]:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
else:
PathOrToken = str(PathOrToken)
share = self.database_get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False)
if share is None:
return httputils.NOT_FOUND
Enabled = None
Hidden = None
if action == "disable":
Enabled = False
elif action == "enable":
Enabled = True
elif action == "hide":
Hidden = True
elif action == "unhide":
Hidden = False
if user == share['Owner']:
if user == share['User']:
# user is Owner and User
result = self.database_update_sharing(
ShareType=ShareType,
PathOrToken=PathOrToken,
EnabledByOwner=Enabled,
EnabledByUser=Enabled,
HiddenByOwner=Hidden,
HiddenByUser=Hidden,
Timestamp=Timestamp)
else:
result = self.database_update_sharing(
ShareType=ShareType,
PathOrToken=PathOrToken,
EnabledByOwner=Enabled,
HiddenByOwner=Hidden,
Timestamp=Timestamp)
elif user == share['User']:
result = self.database_update_sharing(
ShareType=ShareType,
PathOrToken=str(PathOrToken), # verification above that it is not None
EnabledByUser=Enabled,
HiddenByUser=Hidden,
Timestamp=Timestamp)
else:
# neither owner nor user matches
logger.warning("Toggle sharing of %r not permitted for user %r", PathOrToken, user)
return httputils.NOT_ALLOWED
if result:
if result['status'] == "not-found":
return httputils.NOT_FOUND
if result['status'] == "permission-denied":
return httputils.NOT_ALLOWED
elif result['status'] == "success":
answer['Status'] = "success"
pass
else:
logger.error("Toggle sharing: %r of user %s not successful", request_data['PathOrToken'], user)
return httputils.bad_request("Internal Error")
else:
# default
logger.error(api_info + ": unsupported action=%r", action)

250
radicale/sharing/csv.py
View File

@@ -31,7 +31,7 @@ class Sharing(sharing.BaseSharing):
_sharing_db_file: str
# Overloaded functions
def init_database(self) -> bool:
def database_init(self) -> bool:
logger.debug("sharing database initialization for type 'csv'")
sharing_db_file = self.configuration.get("sharing", "database_path")
if sharing_db_file == "":
@@ -76,11 +76,11 @@ class Sharing(sharing.BaseSharing):
self._sharing_db_file = sharing_db_file
return True
def get_database_info(self) -> Union[dict, None]:
def database_get_info(self) -> Union[dict, None]:
database_info = {'type': "csv"}
return database_info
def verify_database(self) -> bool:
def database_verify(self) -> bool:
logger.info("sharing database (csv) verification begin")
logger.info("sharing database (csv) file: %r", self._sharing_db_file)
logger.info("sharing database (csv) loaded entries: %d", self._lines)
@@ -88,14 +88,15 @@ class Sharing(sharing.BaseSharing):
logger.info("sharing database (csv) verification end")
return True
def get_sharing(self,
ShareType: str,
PathOrToken: str,
User: Union[str, None] = None) -> Union[dict, None]:
def database_get_sharing(self,
ShareType: str,
PathOrToken: str,
OnlyEnabled: bool = True,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
# Lookup
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: lookup ShareType=%r PathOrToken=%r User=%r)", ShareType, PathOrToken, User)
logger.debug("TRACE/sharing: lookup ShareType=%r PathOrToken=%r User=%r OnlyEnabled=%s)", ShareType, PathOrToken, User, OnlyEnabled)
index = 0
found = False
@@ -103,25 +104,22 @@ class Sharing(sharing.BaseSharing):
if index == 0:
# skip fieldnames
pass
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: check row: %r", row)
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
elif User is not None and row['User'] != User:
pass
elif row['EnabledByOwner'] is not True:
pass
elif row['ShareType'] == "map":
if row['EnabledByUser'] is not True:
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing: check row: %r", row)
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
elif User is not None and row['User'] != User:
pass
elif OnlyEnabled is True and row['EnabledByOwner'] is not True:
pass
elif OnlyEnabled is True and row['EnabledByUser'] is not True:
pass
else:
found = True
break
else:
found = True
break
index += 1
if found:
@@ -145,16 +143,16 @@ class Sharing(sharing.BaseSharing):
"Properties": Properties}
return None
def list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
def database_list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
""" retrieve sharing """
row: dict
index = 0
@@ -203,15 +201,15 @@ class Sharing(sharing.BaseSharing):
index += 1
return result
def create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
def database_create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
""" create sharing """
row: dict
@@ -266,20 +264,22 @@ class Sharing(sharing.BaseSharing):
logger.error("sharing/%s/create: cannot update CSV database", ShareType)
return {"status": "error"}
def update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
def database_update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: Union[str, None] = None,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
""" update sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r PathMapped=%r Properties=%r", ShareType, PathOrToken, OwnerOrUser, PathMapped, Properties)
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
found = False
@@ -301,24 +301,6 @@ class Sharing(sharing.BaseSharing):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index)
if row['Owner'] != OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r not matching Owner=%r -> check now for matching User=%r", ShareType, OwnerOrUser, row['Owner'], row['User'])
if row['User'] == OwnerOrUser and PathMapped is None and Permissions is None and EnabledByOwner is None and HiddenByOwner is None and Properties is not None:
# user is only permitted to update Properties
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r index=%d is permitted to update Properties", ShareType, OwnerOrUser, PathOrToken, index)
pass
else:
return {"status": "permission-denied"}
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r index=%d", ShareType, OwnerOrUser, PathOrToken, index)
logger.debug("TRACE/sharing/%s/update: orig row[%d]=%r", ShareType, index, row)
# CSV: remove+adjust+readd
if PathMapped is not None:
row["PathMapped"] = PathMapped
@@ -328,8 +310,12 @@ class Sharing(sharing.BaseSharing):
row["User"] = User
if EnabledByOwner is not None:
row["EnabledByOwner"] = EnabledByOwner
if EnabledByUser is not None:
row["EnabledByUser"] = EnabledByUser
if HiddenByOwner is not None:
row["HiddenByOwner"] = HiddenByOwner
if HiddenByUser is not None:
row["HiddenByUser"] = HiddenByUser
if Properties is not None:
row["Properties"] = Properties
# update timestamp
@@ -351,13 +337,12 @@ class Sharing(sharing.BaseSharing):
else:
return {"status": "not-found"}
def delete_sharing(self,
ShareType: str,
PathOrToken: str, Owner: str,
PathMapped: Union[str, None] = None) -> dict:
def database_delete_sharing(self,
ShareType: str,
PathOrToken: str) -> dict:
""" delete sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped)
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r", ShareType, PathOrToken)
# lookup token
found = False
@@ -373,28 +358,18 @@ class Sharing(sharing.BaseSharing):
elif row['PathOrToken'] != PathOrToken:
pass
else:
if ShareType == "map":
# extra filter
if row['PathMapped'] != PathMapped:
pass
else:
found = True
break
else:
found = True
break
found = True
break
index += 1
if found:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: found index=%d", ShareType, index)
if row['Owner'] != Owner:
return {"status": "permission-denied"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index)
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r index=%d", ShareType, PathOrToken, row['Owner'], index)
self._sharing_cache.pop(index)
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
with self._storage.acquire_lock("w", row['Owner'], path=self._sharing_db_file):
if self._write_csv(self._sharing_db_file):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing_by_token: write CSV done")
@@ -404,99 +379,6 @@ class Sharing(sharing.BaseSharing):
else:
return {"status": "not-found"}
def toggle_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
Action: str,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
Timestamp: int = 0) -> dict:
""" toggle sharing """
row: dict
if Action not in sharing.API_SHARE_TOGGLES_V1:
# should not happen
raise
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r", ShareType, Action, OwnerOrUser, User, PathOrToken, PathMapped)
# lookup entry
found = False
index = 0
for row in self._sharing_cache:
if index == 0:
# skip fieldnames
pass
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/*/" + Action + ": check: %r", row)
if row['ShareType'] != ShareType:
pass
elif row['PathOrToken'] != PathOrToken:
pass
elif PathMapped is not None and row['PathMapped'] != PathMapped:
pass
elif row['Owner'] == OwnerOrUser:
found = True
break
else:
found = True
break
index += 1
if found:
# if logger.isEnabledFor(logging.DEBUG):
# logger.debug("TRACE/sharing/*/" + Action + ": found: %r", row)
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
elif row['Owner'] == OwnerOrUser:
pass
elif row['User'] == OwnerOrUser:
pass
else:
return {"status": "permission-denied"}
# TODO: locking
if row['Owner'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: Owner=%r User=%r PathOrToken=%r index=%d", ShareType, Action, OwnerOrUser, User, PathOrToken, index)
if Action == "disable":
row['EnabledByOwner'] = False
elif Action == "enable":
row['EnabledByOwner'] = True
elif Action == "hide":
row['HiddenByOwner'] = True
elif Action == "unhide":
row['HiddenByOwner'] = False
row['TimestampUpdated'] = Timestamp
if row['User'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: User=%r PathOrToken=%r index=%d", ShareType, Action, OwnerOrUser, PathOrToken, index)
if Action == "disable":
row['EnabledByUser'] = False
elif Action == "enable":
row['EnabledByUser'] = True
elif Action == "hide":
row['HiddenByUser'] = True
elif Action == "unhide":
row['HiddenByUser'] = False
row['TimestampUpdated'] = Timestamp
# replace
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 logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE: write CSV done")
return {"status": "success"}
logger.error("sharing: cannot update CSV database")
return {"status": "error"}
else:
return {"status": "not-found"}
# local functions
def _create_empty_csv(self, file: str) -> bool:
with self._storage.acquire_lock("w", None, path=file):

234
radicale/sharing/files.py
View File

@@ -29,19 +29,19 @@ DB_VERSION: str = "1"
class Sharing(sharing.BaseSharing):
_sharing_db_path_ShareType: dict = {}
_sharing_database_path_ShareType: dict = {}
# Overloaded functions
def init_database(self) -> bool:
def database_init(self) -> bool:
logger.debug("sharing database initialization for type 'files'")
sharing_db_path = self.configuration.get("sharing", "database_path")
if sharing_db_path == "":
sharing_database_path = self.configuration.get("sharing", "database_path")
if sharing_database_path == "":
folder = self.configuration.get("storage", "filesystem_folder")
folder_db = os.path.join(folder, "collection-db")
sharing_db_path = os.path.join(folder_db, "files")
logger.info("sharing database path not provided, use default: %r", sharing_db_path)
sharing_database_path = os.path.join(folder_db, "files")
logger.info("sharing database path not provided, use default: %r", sharing_database_path)
else:
logger.info("sharing database path: %r", sharing_db_path)
logger.info("sharing database path: %r", sharing_database_path)
if not os.path.exists(folder_db):
logger.warning("sharing database folder is not existing: %r (create now)", folder_db)
@@ -52,18 +52,18 @@ class Sharing(sharing.BaseSharing):
return False
logger.info("sharing database folder successfully created: %r", folder_db)
if not os.path.exists(sharing_db_path):
logger.warning("sharing database path is not existing: %r", sharing_db_path)
if not os.path.exists(sharing_database_path):
logger.warning("sharing database path is not existing: %r", sharing_database_path)
try:
os.mkdir(sharing_db_path)
os.mkdir(sharing_database_path)
except Exception as e:
logger.error("sharing database path cannot be created (check permissions): %r (%r)", sharing_db_path, e)
logger.error("sharing database path cannot be created (check permissions): %r (%r)", sharing_database_path, e)
return False
logger.info("sharing database path successfully created: %r", sharing_db_path)
logger.info("sharing database path successfully created: %r", sharing_database_path)
for ShareType in sharing.SHARE_TYPES_V1:
path = os.path.join(sharing_db_path, ShareType)
self._sharing_db_path_ShareType[ShareType] = path
path = os.path.join(sharing_database_path, ShareType)
self._sharing_database_path_ShareType[ShareType] = path
if not os.path.exists(path):
logger.warning("sharing database path for %r is not existing: %r", ShareType, path)
try:
@@ -74,28 +74,29 @@ class Sharing(sharing.BaseSharing):
logger.info("sharing database path for %r successfully created: %r", ShareType, path)
return True
def get_database_info(self) -> Union[dict, None]:
def database_get_info(self) -> Union[dict, None]:
database_info = {'type': "files"}
return database_info
def verify_database(self) -> bool:
def database_verify(self) -> bool:
logger.info("sharing database (files) verification begin")
for ShareType in sharing.SHARE_TYPES_V1:
logger.info("sharing database (files) path for %r: %r", ShareType, self._sharing_db_path_ShareType[ShareType])
logger.info("sharing database (files) path for %r: %r", ShareType, self._sharing_database_path_ShareType[ShareType])
# TODO: count amount of files
logger.info("sharing database (files) verification end")
return True
def get_sharing(self,
ShareType: str,
PathOrToken: str,
User: Union[str, None] = None) -> Union[dict, None]:
def database_get_sharing(self,
ShareType: str,
PathOrToken: str,
OnlyEnabled: bool = True,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
# Lookup
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/get: PathOrToken=%r User=%r)", ShareType, PathOrToken, User)
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
sharing_config_file = os.path.join(self._sharing_database_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return None
@@ -111,9 +112,9 @@ class Sharing(sharing.BaseSharing):
if User is not None and row['User'] != User:
return None
elif row['EnabledByOwner'] is not True:
elif OnlyEnabled is True and row['EnabledByOwner'] is not True:
return None
elif row['ShareType'] == "map":
elif OnlyEnabled is True and row['ShareType'] == "map":
if row['EnabledByUser'] is not True:
return None
@@ -140,16 +141,16 @@ class Sharing(sharing.BaseSharing):
return None
def list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
def database_list_sharing(self,
OwnerOrUser: Union[str, None] = None,
ShareType: Union[str, None] = None,
PathOrToken: Union[str, None] = None,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None) -> list[dict]:
""" retrieve sharing """
result = []
@@ -161,7 +162,7 @@ class Sharing(sharing.BaseSharing):
# skip
continue
path = self._sharing_db_path_ShareType[_ShareType]
path = self._sharing_database_path_ShareType[_ShareType]
with self._storage.acquire_lock("r", OwnerOrUser, path=path):
for entry in os.scandir(path):
if not entry.is_file():
@@ -212,19 +213,19 @@ class Sharing(sharing.BaseSharing):
return result
def create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
def database_create_sharing(self,
ShareType: str,
PathOrToken: str, PathMapped: str,
Owner: str, User: str,
Permissions: str = "r",
EnabledByOwner: bool = False, EnabledByUser: bool = False,
HiddenByOwner: bool = True, HiddenByUser: bool = True,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
""" create sharing """
row: dict
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
sharing_config_file = os.path.join(self._sharing_database_path_ShareType[ShareType], self._encode_path(PathOrToken))
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: sharing_config_file=%r", ShareType, sharing_config_file)
@@ -261,22 +262,24 @@ class Sharing(sharing.BaseSharing):
logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
def update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
def database_update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: Union[str, None] = None,
User: Union[str, None] = None,
PathMapped: Union[str, None] = None,
Permissions: Union[str, None] = None,
EnabledByOwner: Union[bool, None] = None,
EnabledByUser: Union[bool, None] = None,
HiddenByOwner: Union[bool, None] = None,
HiddenByUser: Union[bool, None] = None,
Timestamp: int = 0,
Properties: Union[dict, None] = None) -> dict:
""" update sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r User=%r Properties=%r", ShareType, PathOrToken, OwnerOrUser, User, Properties)
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
sharing_config_file = os.path.join(self._sharing_database_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return {"status": "not-found"}
@@ -293,20 +296,6 @@ class Sharing(sharing.BaseSharing):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: check: %r", ShareType, row)
if row['Owner'] != OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r not matching Owner=%r -> check now for matching User=%r", ShareType, OwnerOrUser, row['Owner'], row['User'])
if row['User'] == OwnerOrUser and PathMapped is None and Permissions is None and EnabledByOwner is None and HiddenByOwner is None and Properties is not None:
# user is only permitted to update Properties
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r is permitted to update Properties", ShareType, OwnerOrUser, PathOrToken)
pass
else:
return {"status": "permission-denied"}
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row)
@@ -318,8 +307,12 @@ class Sharing(sharing.BaseSharing):
row["User"] = User
if EnabledByOwner is not None:
row["EnabledByOwner"] = EnabledByOwner
if EnabledByUser is not None:
row["EnabledByUser"] = EnabledByUser
if HiddenByOwner is not None:
row["HiddenByOwner"] = HiddenByOwner
if HiddenByUser is not None:
row["HiddenByUser"] = HiddenByUser
if Properties is not None:
row["Properties"] = Properties
# update timestamp
@@ -338,21 +331,20 @@ class Sharing(sharing.BaseSharing):
logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
def delete_sharing(self,
ShareType: str,
PathOrToken: str, Owner: str,
PathMapped: Union[str, None] = None) -> dict:
def database_delete_sharing(self,
ShareType: str,
PathOrToken: str) -> dict:
""" delete sharing """
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r", ShareType, PathOrToken, Owner)
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r", ShareType, PathOrToken)
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
sharing_config_file = os.path.join(self._sharing_database_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return {"status": "not-found"}
# read content
with self._storage.acquire_lock("r", Owner, path=sharing_config_file):
with self._storage.acquire_lock("r", path=sharing_config_file):
# read file
with open(sharing_config_file, "rb") as fb:
(version, row) = pickle.load(fb)
@@ -360,10 +352,6 @@ class Sharing(sharing.BaseSharing):
if version != DB_VERSION:
return {"status": "error"}
# verify owner
if row['Owner'] != Owner:
return {"status": "permission-denied"}
try:
os.remove(sharing_config_file)
except Exception as e:
@@ -373,88 +361,6 @@ class Sharing(sharing.BaseSharing):
logger.debug("sharing/%s/delete: successful removed share-config: %r", ShareType, sharing_config_file)
return {"status": "success"}
def toggle_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: str,
Action: str,
PathMapped: Union[str, None] = None,
User: Union[str, None] = None,
Timestamp: int = 0) -> dict:
""" toggle sharing """
row: dict
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r", ShareType, Action, OwnerOrUser, User, PathOrToken, PathMapped)
if Action not in sharing.API_SHARE_TOGGLES_V1:
# should not happen
raise
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
if not os.path.isfile(sharing_config_file):
return {"status": "not-found"}
# read content
with self._storage.acquire_lock("w", OwnerOrUser, path=sharing_config_file):
# read file
with open(sharing_config_file, "rb") as fb:
(version, row) = pickle.load(fb)
if version != DB_VERSION:
return {"status": "error"}
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: check: %r", ShareType, Action, row)
# verify ownership or user
if User is not None and row['User'] != User:
return {"status": "permission-denied"}
elif row['Owner'] == OwnerOrUser:
pass
elif row['User'] == OwnerOrUser:
pass
else:
return {"status": "permission-denied"}
if row['Owner'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: Owner=%r User=%r PathOrToken=%r", ShareType, Action, OwnerOrUser, User, PathOrToken)
if Action == "disable":
row['EnabledByOwner'] = False
elif Action == "enable":
row['EnabledByOwner'] = True
elif Action == "hide":
row['HiddenByOwner'] = True
elif Action == "unhide":
row['HiddenByOwner'] = False
row['TimestampUpdated'] = Timestamp
if row['User'] == OwnerOrUser:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/%s: User=%r PathOrToken=%r", ShareType, Action, OwnerOrUser, PathOrToken)
if Action == "disable":
row['EnabledByUser'] = False
elif Action == "enable":
row['EnabledByUser'] = True
elif Action == "hide":
row['HiddenByUser'] = True
elif Action == "unhide":
row['HiddenByUser'] = False
row['TimestampUpdated'] = Timestamp
try:
# write file
with open(sharing_config_file, "wb") as fb:
pickle.dump((version, row), fb)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/%s/create: share-config file stored: %r", ShareType, sharing_config_file)
return {"status": "success"}
except Exception as e:
logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e)
return {"status": "error"}
# local functions
def _encode_path(self, path: str) -> str:
return urllib.parse.quote(path, safe="")

402
radicale/tests/test_sharing.py
View File

@@ -470,7 +470,7 @@ class TestSharingApiSanity(BaseTest):
assert "Status=success" in answer
assert "Lines=1" in answer
logging.info("\n*** disable token#2 (form->text)")
logging.info("\n*** disable token#2 as owner (form->text)")
form_array = ["PathOrToken=" + token2]
_, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status=success" in answer
@@ -483,7 +483,7 @@ class TestSharingApiSanity(BaseTest):
assert answer_dict['Lines'] == 1
assert answer_dict['Content'][0]['EnabledByOwner'] is False
logging.info("\n*** enable token#2 (json->json)")
logging.info("\n*** enable token#2 as owner (json->json)")
json_dict = {}
json_dict['PathOrToken'] = token2
_, headers, answer = self._sharing_api_json("token", "enable", check=200, login="owner:ownerpw", json_dict=json_dict)
@@ -705,6 +705,7 @@ class TestSharingApiSanity(BaseTest):
"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",
@@ -768,38 +769,18 @@ class TestSharingApiSanity(BaseTest):
assert answer_dict['Content'][0]['HiddenByUser'] is True
assert answer_dict['Content'][0]['Permissions'] == "r"
logging.info("\n*** enable map by owner for owner (json->json) -> 403")
logging.info("\n*** enable map by owner (json->json) -> 200")
json_dict = {}
json_dict['User'] = "owner"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "enable", check=403, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** enable map by owner for user (json->json) -> 200")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
logging.info("\n*** enable map by user (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
logging.info("\n*** enable map by user for owner (json->json) -> should fail")
json_dict = {}
json_dict['User'] = "owner"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "enable", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** fetch collection (without credentials)")
_, headers, answer = self.request("GET", path_mapped, check=401)
@@ -874,6 +855,106 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "delete", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user related to user flags (json->json)")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = True
json_dict['Hidden'] = True
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** list as user and check user flags (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, 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]['EnabledByUser'] is True
assert answer_dict['Content'][0]['HiddenByUser'] is True
logging.info("\n*** update map by user related to user flags (json->json)")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = False
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** list as user and check EnabledByUser==False (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, 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]['EnabledByUser'] is False
logging.info("\n*** update map by owner related to owner flags (json->json)")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = False
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** list as user and check owner flags (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, 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]['EnabledByOwner'] is False
assert answer_dict['Content'][0]['HiddenByOwner'] is False
logging.info("\n*** update map by owner related to owner flag Enabled->True (json->json)")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = True
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** list as user and check owner flags (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, 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]['EnabledByOwner'] is True
logging.info("\n*** update map by owner related to Properties -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Properties'] = {"ICAL:calendar-color": "#CCCCCC"}
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** list as user and check Properties (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, 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"
logging.info("\n*** update map by user related to Properties -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Properties'] = {"ICAL:calendar-color": "#DDDDDD"}
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** list as user and check Properties (json->json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, 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"] == "#DDDDDD"
logging.info("\n*** delete map by owner (json->json) -> ok")
json_dict = {}
json_dict['User'] = "user"
@@ -883,6 +964,169 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
def test_sharing_api_map_update_delete_permissions(self) -> None:
"""share-by-map API usage tests."""
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",
"request_content_on_debug": "False"},
"rights": {"type": "owner_only"}})
json_dict: dict
path_shared = "/user/calendarUP-shared-by-owner.ics/"
path_mapped = "/owner/calendarUP.ics/"
path_mapped2 = "/owner/calendarUP2.ics/"
path_mapped_o2 = "/owner2/calendarUP3.ics/"
logging.info("\n*** prepare and test access")
self.mkcalendar(path_mapped, login="owner:ownerpw")
self.mkcalendar(path_mapped2, login="owner:ownerpw")
self.mkcalendar(path_mapped_o2, login="owner2:owner2pw")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
logging.info("\n*** create map with PathMapped and User and PathOrToken (json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
_, 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"
logging.info("\n*** update map by owner: User (json->json) -> 200")
json_dict = {}
json_dict['User'] = "owner"
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: PathMapped (json->json) -> 200")
json_dict = {}
json_dict['PathMapped'] = path_mapped2
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: PathMapped(owner2) (json->json) -> 403")
json_dict = {}
json_dict['PathMapped'] = path_mapped_o2
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: PathOrToken (json->json) -> 404")
json_dict = {}
json_dict['PathOrToken'] = path_mapped
_, headers, answer = self._sharing_api_json("map", "update", check=404, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: Permissions without PathOrToken (json->json) -> 400")
json_dict = {}
json_dict['Permissions'] = "rw"
_, headers, answer = self._sharing_api_json("map", "update", check=400, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: Permissions (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Permissions'] = "rw"
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: Enabled (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = True
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: Hidden (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by owner: Properties (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Properties'] = {}
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by user: Enabled user-mispatch (json->json) -> 403")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = True
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by owner: User (json->json) -> 200")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by user: User (same) (json->json) -> 403")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user: PathMapped (json->json) -> 403")
json_dict = {}
json_dict['PathMapped'] = path_shared
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user: PathOrToken (json->json) -> 404")
json_dict = {}
json_dict['PathOrToken'] = path_mapped
_, headers, answer = self._sharing_api_json("map", "update", check=404, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user: Permissions (json->json) -> 403")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Permissions'] = "rw"
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user: Enabled (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Enabled'] = True
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user: Hidden (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** update map by user: Properties (json->json) -> 200")
json_dict = {}
json_dict['PathOrToken'] = path_shared
json_dict['Properties'] = {}
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** delete map by user (json->json) -> 403")
json_dict = {}
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "delete", check=403, login="user:userpw", json_dict=json_dict)
logging.info("\n*** delete map by owner (json->json) -> ok")
json_dict = {}
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "delete", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
logging.info("\n*** delete map by owner 2nd time (json->json) -> 404")
json_dict = {}
json_dict['PathOrToken'] = path_shared
_, headers, answer = self._sharing_api_json("map", "delete", check=404, login="owner:ownerpw", json_dict=json_dict)
def test_sharing_api_map_usercheck(self) -> None:
"""share-by-map API usage tests related to usercheck."""
self.configure({"auth": {"type": "htpasswd",
@@ -954,12 +1198,12 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
logging.info("\n*** create map user2/owner1 -> fail")
logging.info("\n*** create map user2/owner1 -> 409 (conflict)")
json_dict = {}
json_dict['User'] = "user2"
json_dict['PathMapped'] = path_mapped2
json_dict['PathOrToken'] = path_share1
_, headers, answer = self._sharing_api_json("map", "create", check=403, login="owner2:owner2pw", json_dict=json_dict)
_, headers, answer = self._sharing_api_json("map", "create", check=409, login="owner2:owner2pw", json_dict=json_dict)
logging.info("\n*** delete map user1 -> ok")
json_dict = {}
@@ -1012,13 +1256,22 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self.request("GET", path_mapped + "event1.ics", check=200, login="owner:ownerpw")
# create maps
logging.info("\n*** create map user/owner:r -> ok")
logging.info("\n*** create map user/owner:r -> 400 (Enabled is not boolean)")
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"
_, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict)
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
_, 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"
@@ -1029,7 +1282,7 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_w
json_dict['Permissions'] = "w"
json_dict['Enabled'] = "True"
json_dict['Enabled'] = True
_, 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"
@@ -1040,7 +1293,7 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_rw
json_dict['Permissions'] = "rw"
json_dict['Enabled'] = "True"
json_dict['Enabled'] = True
_, 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"
@@ -1063,22 +1316,16 @@ class TestSharingApiSanity(BaseTest):
# enable maps by user
logging.info("\n*** enable map by user:r")
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)
logging.info("\n*** enable map by user:w")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_w
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** enable map by user:rw")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_rw
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
@@ -1226,8 +1473,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1337,8 +1584,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1470,8 +1717,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1589,8 +1836,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1601,8 +1848,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_w
json_dict['Permissions'] = "w"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1613,8 +1860,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_rw
json_dict['Permissions'] = "rw"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1751,8 +1998,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1763,8 +2010,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_rw
json_dict['Permissions'] = "rw"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1775,8 +2022,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped2
json_dict['PathOrToken'] = path_shared2_rw
json_dict['Permissions'] = "rw"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -1922,8 +2169,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1
json_dict['Permissions'] = "w"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -2061,8 +2308,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_owner1
json_dict['PathOrToken'] = path_user1_shared1
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
@@ -2088,8 +2335,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_owner2
json_dict['PathOrToken'] = path_user1_shared2
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner2:owner2pw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
@@ -2108,8 +2355,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_owner1
json_dict['PathOrToken'] = path_user2_shared1
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
@@ -2182,8 +2429,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_owner1
json_dict['PathOrToken'] = path_user1_shared1
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
@@ -2198,8 +2445,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_owner1
json_dict['PathOrToken'] = path_user2_shared1
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
@@ -2214,8 +2461,8 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathMapped'] = path_owner1
json_dict['PathOrToken'] = path_user2
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
json_dict['Enabled'] = True
json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=409, login="owner1:owner1pw", json_dict=json_dict)
def test_sharing_api_permissions_global(self) -> None:
@@ -2637,8 +2884,8 @@ permissions: RrWw""")
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -2668,7 +2915,6 @@ permissions: RrWw""")
# update map by user
logging.info("\n*** update map by user (json)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared_r
json_dict['Properties'] = {"C:calendar-description": "ICAL-USER", "ICAL:calendar-color": "#BBBBBB"}
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
@@ -2718,7 +2964,7 @@ permissions: RrWw""")
# update map by user
logging.info("\n*** update map by user (form)")
form_array = ["User=" + "user"]
form_array = []
form_array.append("PathOrToken=" + path_shared_r)
form_array.append("Properties='C:calendar-description'='ICAL-USER-NEW'")
form_array.append("Properties='ICAL:calendar-color'='#CCCCCC'")
@@ -2807,8 +3053,8 @@ permissions: RrWw""")
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"
@@ -2853,7 +3099,6 @@ permissions: RrWw""")
# update map by user
logging.info("\n*** update map by user (json) -> 403 (no overlay permitted)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared_r
json_dict['Properties'] = {"ICAL:calendar-color": "#BBBBBB"}
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
@@ -2871,7 +3116,6 @@ permissions: RrWw""")
# update map by owner
logging.info("\n*** update map by owner (disable property overlay)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "rp"
@@ -2891,7 +3135,6 @@ permissions: RrWw""")
logging.info("\n*** update map by user (json) -> 403 (no overlay permitted by share permissions)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared_r
json_dict['Properties'] = {"ICAL:calendar-color": "#CCCCCC"}
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
@@ -2904,7 +3147,6 @@ permissions: RrWw""")
# update map by owner
logging.info("\n*** update map by owner (disable property overlay)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "rP"
@@ -2913,7 +3155,6 @@ permissions: RrWw""")
logging.info("\n*** update map by user (json) -> 200 (overlay permitted by share permissions)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared_r
json_dict['Properties'] = {"ICAL:calendar-color": "#CCCCCC"}
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
@@ -2928,16 +3169,13 @@ permissions: RrWw""")
# update map by owner
logging.info("\n*** update map by owner (disable property overlay)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "rp"
json_dict['User'] = "user"
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict)
logging.info("\n*** update map by user (json) -> 403 (overlay permitted but denied by share permissions)")
json_dict = {}
json_dict['User'] = "user"
json_dict['PathOrToken'] = path_shared_r
json_dict['Properties'] = {"ICAL:calendar-color": "#EEEEEE"}
_, headers, answer = self._sharing_api_json("map", "update", check=403, login="user:userpw", json_dict=json_dict)
@@ -3003,8 +3241,8 @@ permissions: RrWw""")
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
json_dict['Permissions'] = "rP"
json_dict['Enabled'] = "True"
json_dict['Hidden'] = "False"
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"