sharing: review enable/disable/hide/unhide API

This commit is contained in:
Peter Bieringer
2026-03-03 17:39:00 +01:00
parent 3267cc5228
commit c74365cd8a
5 changed files with 92 additions and 243 deletions

View File

@@ -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)

View File

@@ -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)

View File

@@ -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):

View File

@@ -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="")

View File

@@ -470,7 +470,7 @@ class TestSharingApiSanity(BaseTest):
assert "Status=success" in answer
assert "Lines=1" in answer
logging.info("\n*** disable token#2 (form->text)")
logging.info("\n*** disable token#2 as 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",