review update+delete

This commit is contained in:
Peter Bieringer
2026-03-03 08:34:05 +01:00
parent f9399dc17e
commit 3267cc5228
5 changed files with 399 additions and 155 deletions

View File

@@ -36,6 +36,7 @@ INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none")
DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated', 'Properties')
DB_FIELDS_V1_BOOL: Sequence[str] = ('EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser')
DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
DB_FIELDS_V1_USER_PERMITTED: Sequence[str] = ('EnabledByUser', 'HiddenByUser', 'Properties')
# ShareType: <token|map>
# PathOrToken: <path|token> [PrimaryKey]
# PathMapped: <path>
@@ -48,6 +49,7 @@ DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
# HiddenByUser: True|False (share exposure controlled by user) - check skipped if Owner==User
# TimestampCreated: <unixtime> (when created)
# TimestampUpdated: <unixtime> (last update)
# Properties: Overlay of collection properties
SHARE_TYPES: Sequence[str] = ('token', 'map', 'all')
SHARE_TYPES_V1: Sequence[str] = ('token', 'map')
@@ -164,15 +166,16 @@ class BaseSharing:
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]:
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,
OnlyEnabled: bool = True,
User: Union[str, None] = None) -> Union[dict, None]:
""" retrieve sharing target and attributes by map """
return {"status": "not-implemented"}
@@ -192,12 +195,14 @@ class BaseSharing:
def update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: 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 """
@@ -205,9 +210,7 @@ class BaseSharing:
def delete_sharing(self,
ShareType: str,
PathOrToken: str,
Owner: str,
PathMapped: Union[str, None] = None) -> dict:
PathOrToken: str) -> dict:
""" delete sharing """
return {"status": "not-implemented"}
@@ -504,6 +507,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 == "Enabled" or key == "Hidden":
try:
request_data[key] = config._convert_to_bool(request_parsed[key])
except ValueError:
logger.error(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for " + key)
else:
request_data[key] = request_parsed[key][0]
if logger.isEnabledFor(logging.DEBUG):
@@ -566,10 +575,6 @@ 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)
@@ -577,7 +582,7 @@ class BaseSharing:
# check for mandatory parameters
if 'PathMapped' not in request_data:
if action in ['info', 'list', 'update']:
if action in ['info', 'list', 'update', 'delete']:
# ignored
pass
else:
@@ -617,6 +622,16 @@ class BaseSharing:
return httputils.bad_request("Property not supported to overlay: %r" % entry)
Properties = request_data['Properties']
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 ShareType == "map":
if action == 'info':
# ignored
@@ -809,66 +824,72 @@ 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")
if PathOrToken is None:
return httputils.bad_request("Missing PathOrToken")
# check for permissions to update
share = self.get_sharing(ShareType=ShareType, PathOrToken=PathOrToken)
share = self.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
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.update_sharing(
ShareType=ShareType,
PathMapped=PathMapped,
Permissions=Permissions,
EnabledByOwner=EnabledByOwner,
HiddenByOwner=HiddenByOwner,
EnabledByOwner=Enabled,
HiddenByOwner=Hidden,
PathOrToken=str(PathOrToken), # verification above that it is not None
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 EnabledByOwner is not None or HiddenByOwner 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(
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,25 +912,26 @@ class BaseSharing:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/" + api_info + ": start")
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")
if ShareType == "token":
# check whether share exists
share = self.get_sharing(ShareType=ShareType, PathOrToken=PathOrToken, OnlyEnabled=False)
if share is None:
return httputils.NOT_FOUND
if user == share['Owner']:
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)
PathOrToken=str(PathOrToken)) # verification above that it is not None
else:
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
return httputils.bad_request("Invalid share type")
# 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":

View File

@@ -91,6 +91,7 @@ class Sharing(sharing.BaseSharing):
def 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
@@ -103,25 +104,26 @@ 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 row['ShareType'] == "map":
if OnlyEnabled is True and row['EnabledByUser'] is not True:
pass
else:
found = True
break
else:
found = True
break
else:
found = True
break
index += 1
if found:
@@ -269,17 +271,19 @@ class Sharing(sharing.BaseSharing):
def update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: 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 +305,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 +314,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
@@ -353,11 +343,10 @@ class Sharing(sharing.BaseSharing):
def delete_sharing(self,
ShareType: str,
PathOrToken: str, Owner: str,
PathMapped: Union[str, None] = None) -> dict:
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 +362,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")

View File

@@ -89,6 +89,7 @@ class Sharing(sharing.BaseSharing):
def 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
@@ -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
@@ -264,12 +265,14 @@ class Sharing(sharing.BaseSharing):
def update_sharing(self,
ShareType: str,
PathOrToken: str,
OwnerOrUser: 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 """
@@ -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
@@ -340,11 +333,10 @@ class Sharing(sharing.BaseSharing):
def delete_sharing(self,
ShareType: str,
PathOrToken: str, Owner: str,
PathMapped: Union[str, None] = None) -> dict:
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))
@@ -352,7 +344,7 @@ class Sharing(sharing.BaseSharing):
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: