diff --git a/SHARING.md b/SHARING.md index d2c290af..ef413558 100644 --- a/SHARING.md +++ b/SHARING.md @@ -245,6 +245,10 @@ map,/user/cal1-from-owner/,/owner/cal1/,owner,user,r,False,False,True,True,17719 Delete a share selected by `PathOrToken`. + * Authorization + +Authenticated user is `Owner` + * Input | Parameter | Type | Owner | User | @@ -259,6 +263,10 @@ 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 | Type | Owner | User | @@ -274,17 +282,21 @@ Execute delete+create in case `PathOrToken` needs to be changed. * 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/sharing/__init__.py b/radicale/sharing/__init__.py index 428e4555..8adc4b9f 100644 --- a/radicale/sharing/__init__.py +++ b/radicale/sharing/__init__.py @@ -214,17 +214,6 @@ class BaseSharing: """ 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 """ @@ -582,7 +571,7 @@ class BaseSharing: # check for mandatory parameters if 'PathMapped' not in request_data: - if action in ['info', 'list', 'update', 'delete']: + if action in ['info', 'list', 'update', 'delete', 'enable', 'disable', 'hide', 'unhide']: # ignored pass else: @@ -638,7 +627,7 @@ class BaseSharing: pass else: if 'User' not in request_data: - if action not in ['list', 'delete', 'update']: + if action not in ['list', 'delete', 'update', 'enable', 'disable', 'hide', 'unhide']: logger.warning(api_info + ": missing User") return httputils.bad_request("Missing User") else: @@ -963,35 +952,73 @@ 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") + + share = self.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.update_sharing( + ShareType=ShareType, + PathOrToken=str(PathOrToken), # verification above that it is not None + EnabledByOwner=Enabled, + EnabledByUser=Enabled, + HiddenByOwner=Hidden, + HiddenByUser=Hidden, + Timestamp=Timestamp) + else: + result = self.update_sharing( + ShareType=ShareType, + PathOrToken=str(PathOrToken), # verification above that it is not None + EnabledByOwner=Enabled, + HiddenByOwner=Hidden, + Timestamp=Timestamp) + + elif user == share['User']: + result = self.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 392eb385..b5c01a53 100644 --- a/radicale/sharing/csv.py +++ b/radicale/sharing/csv.py @@ -383,99 +383,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 e7b16875..1f3ebed7 100644 --- a/radicale/sharing/files.py +++ b/radicale/sharing/files.py @@ -361,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 8a57159b..33d2f7a6 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 iwner (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) @@ -769,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) @@ -1142,6 +1122,11 @@ class TestSharingApiSanity(BaseTest): 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",