diff --git a/SHARING.md b/SHARING.md index 4ec3e49d..daebf668 100644 --- a/SHARING.md +++ b/SHARING.md @@ -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) diff --git a/radicale/app/proppatch.py b/radicale/app/proppatch.py index 1f1dc5f6..59d56c83 100644 --- a/radicale/app/proppatch.py +++ b/radicale/app/proppatch.py @@ -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 diff --git a/radicale/sharing/__init__.py b/radicale/sharing/__init__.py index 08ac554e..45e05ed1 100644 --- a/radicale/sharing/__init__.py +++ b/radicale/sharing/__init__.py @@ -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: # PathOrToken: [PrimaryKey] # PathMapped: @@ -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: (when created) # TimestampUpdated: (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) diff --git a/radicale/sharing/csv.py b/radicale/sharing/csv.py index ea0d8c99..cf761923 100644 --- a/radicale/sharing/csv.py +++ b/radicale/sharing/csv.py @@ -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): diff --git a/radicale/sharing/files.py b/radicale/sharing/files.py index 1e584b5c..3a79730f 100644 --- a/radicale/sharing/files.py +++ b/radicale/sharing/files.py @@ -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="") diff --git a/radicale/tests/test_sharing.py b/radicale/tests/test_sharing.py index ec17b163..d61d1b2c 100644 --- a/radicale/tests/test_sharing.py +++ b/radicale/tests/test_sharing.py @@ -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"