diff --git a/SHARING.md b/SHARING.md index 62750c61..ce857a77 100644 --- a/SHARING.md +++ b/SHARING.md @@ -28,6 +28,7 @@ Types of supported sharing configuration: * `HiddenByUser`: control by user * `TimestampCreated`: unixtime of creation * `TimestampUpdated`: unixtime of last update + * `Properties`: overlay properties (limited set whitelisted) `Enabled*`: owner AND user have to enable a share to become usable @@ -39,7 +40,9 @@ Types of supported sharing configuration: (_>= 3.7.0_) -One CSV file containing one row per sharing config, separated by `,` and containing header with columns from above. +One CSV file containing one row per sharing config, separated by `;` and containing header with columns from above. + +If given, properties are stored in JSON format in CSV. #### Files @@ -120,6 +123,7 @@ Can be selected by `HTTP_ACCEPT` * `Permissions`: effective permission of the share * `Enabled`: owner/user selected by authentication * `Hidden`: owner/user selected by authentication + * `Properties`: properties to overlay #### API Hooks @@ -250,7 +254,9 @@ Update a share selected by `PathOrToken` | - | - | - | | PathOrToken | yes | n/a | | PathMapped | no | | +| OwnerOrUser | yes | n/a | | User | no | | +| Properties | no | | * Output: result status @@ -282,4 +288,13 @@ ApiVersion=1 Status=success ``` +## Properties Overlay +Owner or user can define per share a set of properties to overlay on PROPFIND response during create or update via API. + +Whitelisted ones are defined in `OVERLAY_PROPERTIES_WHITELIST` in `radicale/sharing/__init__.py`: + +* `C:calendar-description` (_>= 3.7.0_) +* `ICAL:calendar-color` (_>= 3.7.0_) +* `CR:addressbook-description` (_>= 3.7.0_) +* `INF:addressbook-color` (_>= 3.7.0_) diff --git a/radicale/app/propfind.py b/radicale/app/propfind.py index fc797c74..96a32eae 100644 --- a/radicale/app/propfind.py +++ b/radicale/app/propfind.py @@ -345,6 +345,10 @@ def xml_propfind_response( human_tag = xmlutils.make_human_tag(tag) tag_text = collection.get_meta(human_tag) if tag_text is not None: + if sharing: + # map from overlay + if sharing['Properties'][human_tag] is not None: + tag_text = sharing['Properties'][human_tag] element.text = tag_text else: is404 = True diff --git a/radicale/sharing/__init__.py b/radicale/sharing/__init__.py index d96d3d7d..76ed8e24 100644 --- a/radicale/sharing/__init__.py +++ b/radicale/sharing/__init__.py @@ -33,7 +33,7 @@ from radicale.log import logger INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none") -DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated') +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') # ShareType: @@ -76,6 +76,8 @@ PATH_PATTERN: str = "([a-zA-Z0-9/.\\-]+)" # TODO: extend or find better source USER_PATTERN: str = "([a-zA-Z0-9@]+)" # TODO: extend or find better source +OVERLAY_PROPERTIES_WHITELIST: Sequence[str] = ("C:calendar-description", "ICAL:calendar-color", "CR:addressbook-description", "INF:addressbook-color") + def load(configuration: "config.Configuration") -> "BaseSharing": """Load the sharing database module chosen in configuration.""" @@ -178,20 +180,22 @@ class BaseSharing: Permissions: str = "r", EnabledByOwner: bool = False, EnabledByUser: bool = False, HiddenByOwner: bool = True, HiddenByUser: bool = True, - Timestamp: int = 0) -> dict: + Timestamp: int = 0, + Properties: Union[dict, None] = None) -> dict: """ create sharing """ return {"status": "not-implemented"} def update_sharing(self, ShareType: str, PathOrToken: str, - Owner: Union[str, None] = None, + 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) -> dict: + Timestamp: int = 0, + Properties: Union[dict, None] = None) -> dict: """ update sharing """ return {"status": "not-implemented"} @@ -391,6 +395,8 @@ class BaseSharing: PathOrToken: (mandatory) User: (mandatory) + action: (token|map)/update + action: (token|map)/(delete|disable|enable|hide|unhide) PathOrToken: (mandatory) @@ -408,7 +414,7 @@ class BaseSharing: Status in JSON/TEXT (TEXT can be parsed by shell) """ - if not self.sharing_collection_by_map and not self.sharing_collection_by_token: + if not self._enabled: # API is not enabled return httputils.NOT_FOUND @@ -420,7 +426,7 @@ class BaseSharing: if not path.startswith("/.sharing/v1/"): return httputils.NOT_FOUND - # split into ShareType and action or "info" + # split into ShareType and action ShareType_action = path.removeprefix("/.sharing/v1/") match = re.search('([a-z]+)/([a-z]+)$', ShareType_action) if not match: @@ -481,7 +487,21 @@ class BaseSharing: # convert arrays into single value request_data = {} for key in request_parsed: - request_data[key] = request_parsed[key][0] + if key == "Properties": + # Properties key value parser + properties_dict: dict = {} + for entry in request_parsed[key]: + m = re.search('^([^=]+)=([^=]+)$', entry) + if not m: + return httputils.bad_request("Invalid properties format in form") + token = m.group(1).lstrip('"\'').rstrip('"\'') + value = m.group(2).lstrip('"\'').rstrip('"\'') + properties_dict[token] = value + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/API: converted Properties from form into dict: %r", properties_dict) + request_data[key] = properties_dict + else: + request_data[key] = request_parsed[key][0] if logger.isEnabledFor(logging.DEBUG): logger.debug("TRACE/" + api_info + " (form): %r", f"{request_data}") else: @@ -518,6 +538,7 @@ class BaseSharing: HiddenByOwner: Union[bool, None] = None EnabledByUser: Union[bool, None] = None HiddenByUser: Union[bool, None] = None + Properties: Union[dict, None] = None # parameters sanity check for key in request_data: @@ -552,12 +573,9 @@ class BaseSharing: # check for mandatory parameters if 'PathMapped' not in request_data: - if action == 'info': + if action in ['info', 'list', 'update']: # ignored pass - elif action == "list": - # optional - pass else: if ShareType == "token" and action != 'create': # optional @@ -588,6 +606,13 @@ class BaseSharing: if 'Permissions' in request_data: Permissions = request_data['Permissions'] + if 'Properties' in request_data: + # verify against whitelist + for entry in request_data['Properties']: + if entry not in OVERLAY_PROPERTIES_WHITELIST: + return httputils.bad_request("Property not supported to overlay: %r" % entry) + Properties = request_data['Properties'] + if ShareType == "map": if action == 'info': # ignored @@ -609,6 +634,11 @@ class BaseSharing: answer['ApiVersion'] = 1 Timestamp = int((datetime.now() - datetime(1970, 1, 1)).total_seconds()) + if not self.sharing_collection_by_map and not self.sharing_collection_by_token: + if not action == 'info': + # API is not enabled + return httputils.NOT_FOUND + # action: list if action == "list": if logger.isEnabledFor(logging.DEBUG): @@ -691,7 +721,8 @@ class BaseSharing: Owner=Owner, User=Owner, Permissions=str(Permissions), # mandantory EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner, - Timestamp=Timestamp) + Timestamp=Timestamp, + Properties=Properties) if logger.isEnabledFor(logging.DEBUG): logger.debug("TRACE/" + api_info + ": result=%r", result) @@ -747,7 +778,8 @@ class BaseSharing: Permissions=str(Permissions), # mandatory EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner, EnabledByUser=EnabledByUser, HiddenByUser=HiddenByUser, - Timestamp=Timestamp) + Timestamp=Timestamp, + Properties=Properties) else: logger.error(api_info + ": unsupported for ShareType=%r", ShareType) @@ -784,8 +816,10 @@ class BaseSharing: EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner, PathOrToken=str(PathOrToken), # verification above that it is not None - Owner=Owner, - Timestamp=Timestamp) + OwnerOrUser=Owner, + User=User, + Timestamp=Timestamp, + Properties=Properties) elif ShareType == "map": result = self.update_sharing( @@ -795,8 +829,10 @@ class BaseSharing: EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner, PathOrToken=str(PathOrToken), # verification above that it is not None - Owner=Owner, - Timestamp=Timestamp) + OwnerOrUser=Owner, + User=User, + Timestamp=Timestamp, + Properties=Properties) else: logger.error(api_info + ": unsupported for ShareType=%r", ShareType) @@ -918,18 +954,19 @@ class BaseSharing: answer_array.append(key + '=' + str(answer[key])) if 'Content' in answer and answer['Content'] is not None: csv = io.StringIO() - writer = DictWriter(csv, fieldnames=DB_FIELDS_V1) + writer = DictWriter(csv, fieldnames=DB_FIELDS_V1, delimiter=';') if output_format == "csv": writer.writeheader() for entry in answer['Content']: - writer.writerow(entry) + # TODO: Argument 1 to "writerow" of "DictWriter" has incompatible type "str"; expected "Mapping[str, Any]" [arg-type] + writer.writerow(entry) # type: ignore[arg-type] if output_format == "csv": answer_array.append(csv.getvalue()) else: index = 0 for line in csv.getvalue().splitlines(): # create a shell array with content lines - answer_array.append('Content[' + str(index) + ']="' + line + '"') + answer_array.append('Content[' + str(index) + ']="' + line.replace('"', '\\"') + '"') index += 1 headers = { "Content-Type": "text/csv" diff --git a/radicale/sharing/csv.py b/radicale/sharing/csv.py index b0481d55..1723c495 100644 --- a/radicale/sharing/csv.py +++ b/radicale/sharing/csv.py @@ -128,8 +128,9 @@ class Sharing(sharing.BaseSharing): UserShare = row['User'] Permissions = row['Permissions'] Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser']) - if logger.isEnabledFor(logging.DEBUG): - logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden) + Properties: Union[dict, None] = None + if 'Properties' in row: + Properties = row['Properties'] return { "mapped": True, "PathOrToken": PathOrToken, @@ -137,7 +138,8 @@ class Sharing(sharing.BaseSharing): "Owner": Owner, "User": UserShare, "Hidden": Hidden, - "Permissions": Permissions} + "Permissions": Permissions, + "Properties": Properties} return None def list_sharing(self, @@ -205,7 +207,8 @@ class Sharing(sharing.BaseSharing): Permissions: str = "r", EnabledByOwner: bool = False, EnabledByUser: bool = False, HiddenByOwner: bool = True, HiddenByUser: bool = True, - Timestamp: int = 0) -> dict: + Timestamp: int = 0, + Properties: Union[dict, None] = None) -> dict: """ create sharing """ row: dict @@ -263,16 +266,17 @@ class Sharing(sharing.BaseSharing): def update_sharing(self, ShareType: str, PathOrToken: str, - Owner: Union[str, None] = None, + 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) -> dict: + Timestamp: int = 0, + Properties: Union[dict, None] = None) -> dict: """ update sharing """ if logger.isEnabledFor(logging.DEBUG): - logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped) + logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r PathMapped=%r Properties=%r", ShareType, PathOrToken, OwnerOrUser, PathMapped, Properties) # lookup token found = False @@ -293,12 +297,23 @@ class Sharing(sharing.BaseSharing): if found: if logger.isEnabledFor(logging.DEBUG): logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index) - if Owner is not None and row['Owner'] != Owner: - return {"status": "permission-denied"} + + 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: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index) + logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r index=%d", ShareType, OwnerOrUser, PathOrToken, index) logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row) # CSV: remove+adjust+readd @@ -312,6 +327,8 @@ class Sharing(sharing.BaseSharing): row["EnabledByOwner"] = EnabledByOwner if HiddenByOwner is not None: row["HiddenByOwner"] = HiddenByOwner + if Properties is not None: + row["Properties"] = Properties # update timestamp row["TimestampUpdated"] = Timestamp @@ -322,7 +339,7 @@ class Sharing(sharing.BaseSharing): self._sharing_cache.pop(index) self._sharing_cache.append(row) - with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file): + 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/sharing/%s/update: write CSV done", ShareType) @@ -484,7 +501,7 @@ class Sharing(sharing.BaseSharing): def _create_empty_csv(self, file: str) -> bool: with self._storage.acquire_lock("w", None, path=file): with open(file, 'w', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1) + writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';') writer.writeheader() return True @@ -492,7 +509,7 @@ class Sharing(sharing.BaseSharing): logger.debug("sharing database load begin: %r", file) with self._storage.acquire_lock("r", None): with open(file, 'r', newline='') as csvfile: - reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1) + reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';') self._lines = 0 for row in reader: # logger.debug("sharing database load read: %r", row) @@ -506,7 +523,10 @@ class Sharing(sharing.BaseSharing): # convert txt to bool if self._lines > 0: for fieldname in sharing.DB_FIELDS_V1_BOOL: - row[fieldname] = config._convert_to_bool(row[fieldname]) + try: + row[fieldname] = config._convert_to_bool(row[fieldname]) + except Exception as e: + logger.error("sharing database row error fieldname=%r row=%r error: %r", fieldname, row, e) for fieldname in sharing.DB_FIELDS_V1_INT: row[fieldname] = int(row[fieldname]) # check for duplicates @@ -525,6 +545,6 @@ class Sharing(sharing.BaseSharing): def _write_csv(self, file: str) -> bool: with open(file, 'w', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1) + writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';') writer.writerows(self._sharing_cache) return True diff --git a/radicale/sharing/files.py b/radicale/sharing/files.py index e268099b..40dcb7b2 100644 --- a/radicale/sharing/files.py +++ b/radicale/sharing/files.py @@ -122,8 +122,11 @@ class Sharing(sharing.BaseSharing): UserShare = row['User'] Permissions = row['Permissions'] Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser']) + Properties: Union[dict, None] = None + if 'Properties' in row: + Properties = row['Properties'] if logger.isEnabledFor(logging.DEBUG): - logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden) + logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s Properties=%r)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden, Properties) return { "mapped": True, "PathOrToken": PathOrToken, @@ -131,7 +134,8 @@ class Sharing(sharing.BaseSharing): "Owner": Owner, "User": UserShare, "Hidden": Hidden, - "Permissions": Permissions} + "Permissions": Permissions, + "Properties": Properties} return None @@ -214,7 +218,8 @@ class Sharing(sharing.BaseSharing): Permissions: str = "r", EnabledByOwner: bool = False, EnabledByUser: bool = False, HiddenByOwner: bool = True, HiddenByUser: bool = True, - Timestamp: int = 0) -> dict: + Timestamp: int = 0, + Properties: Union[dict, None] = None) -> dict: """ create sharing """ row: dict @@ -258,16 +263,17 @@ class Sharing(sharing.BaseSharing): def update_sharing(self, ShareType: str, PathOrToken: str, - Owner: Union[str, None] = None, + 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) -> dict: + Timestamp: int = 0, + Properties: Union[dict, None] = None) -> dict: """ update sharing """ if logger.isEnabledFor(logging.DEBUG): - logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r User=%r", ShareType, PathOrToken, Owner, User) + 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)) @@ -275,7 +281,7 @@ class Sharing(sharing.BaseSharing): return {"status": "not-found"} # read content - with self._storage.acquire_lock("w", Owner, path=sharing_config_file): + 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) @@ -286,8 +292,17 @@ class Sharing(sharing.BaseSharing): if logger.isEnabledFor(logging.DEBUG): logger.debug("TRACE/sharing/%s/update: check: %r", ShareType, row) - if Owner is not None and row['Owner'] != Owner: - return {"status": "permission-denied"} + 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"} @@ -304,6 +319,8 @@ class Sharing(sharing.BaseSharing): row["EnabledByOwner"] = EnabledByOwner if HiddenByOwner is not None: row["HiddenByOwner"] = HiddenByOwner + if Properties is not None: + row["Properties"] = Properties # update timestamp row["TimestampUpdated"] = Timestamp diff --git a/radicale/tests/test_sharing.py b/radicale/tests/test_sharing.py index 2cf9f508..a15dceae 100644 --- a/radicale/tests/test_sharing.py +++ b/radicale/tests/test_sharing.py @@ -76,25 +76,56 @@ class TestSharingApiSanity(BaseTest): # disabled for path in ["/.sharing", "/.sharing/"]: _, headers, _ = self.request("POST", path, check=404) - # enabled (permutations) + + path = "/.sharing/" + + # no database is active + logging.info("\n*** check API hook base: map=True token=False") self.configure({"sharing": { "collection_by_map": "True", "collection_by_token": "False"} }) - path = "/.sharing/" - _, headers, _ = self.request("POST", path, check=401) + _, headers, _ = self.request("POST", path, check=404) + + logging.info("\n*** check API hook base: map=False token=True") self.configure({"sharing": { "collection_by_map": "False", "collection_by_token": "True"} }) - path = "/.sharing/" - _, headers, _ = self.request("POST", path, check=401) + _, headers, _ = self.request("POST", path, check=404) + + logging.info("\n*** check API hook base: map=True token=True") self.configure({"sharing": { "collection_by_map": "True", "collection_by_token": "True"} }) - path = "/.sharing/" - _, headers, _ = self.request("POST", path, check=401) + _, headers, _ = self.request("POST", path, check=404) + + 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}}) + + # no database is active + logging.info("\n*** check API hook base: map=True token=False") + self.configure({"sharing": { + "collection_by_map": "True", + "collection_by_token": "False"} + }) + _, headers, _ = self.request("POST", path, check=401) + + logging.info("\n*** check API hook base: map=False token=True") + self.configure({"sharing": { + "collection_by_map": "False", + "collection_by_token": "True"} + }) + _, headers, _ = self.request("POST", path, check=401) + + logging.info("\n*** check API hook base: map=True token=True") + self.configure({"sharing": { + "collection_by_map": "True", + "collection_by_token": "True"} + }) + _, headers, _ = self.request("POST", path, check=401) def test_sharing_api_base_with_auth(self) -> None: """POST request at '/.sharing' with authentication.""" @@ -108,75 +139,82 @@ class TestSharingApiSanity(BaseTest): json_dict: dict - # path with no valid API hook - for path in ["/.sharing/", "/.sharing/v9/"]: - _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") + 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}}) - # path with valid API but no hook - for path in ["/.sharing/v1/"]: - _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") + for path in ["/.sharing/", "/.sharing/v9/"]: + logging.info("\n*** check invalid API URI: %r", path) + _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") - # path with valid API and hook but not enabled "map" - self.configure({"sharing": { - "collection_by_map": "False", - "collection_by_token": "True"} - }) - sharetype = "map" - for action in sharing.API_HOOKS_V1: - path = "/.sharing/v1/" + sharetype + "/" + action - _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") + # path with valid API but no hook + for path in ["/.sharing/v1/"]: + logging.info("\n*** check valid API URI without hook: %r", path) + _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") - # path with valid API and hook but not enabled "token" - self.configure({"sharing": { - "collection_by_map": "True", - "collection_by_token": "False"} - }) - sharetype = "token" - for action in sharing.API_HOOKS_V1: - path = "/.sharing/v1/" + sharetype + "/" + action - _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") + # path with valid API and hook but not enabled "map" + self.configure({"sharing": { + "collection_by_map": "False", + "collection_by_token": "True"} + }) + sharetype = "map" + for action in sharing.API_HOOKS_V1: + path = "/.sharing/v1/" + sharetype + "/" + action + logging.info("\n*** check valid API URI hook (but not enabled): %r", path) + _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") - # check info hook - logging.info("\n*** check API hook: info/all") - json_dict = {} - _, headers, answer = self._sharing_api_json("all", "info", check=200, login="owner:ownerpw", json_dict=json_dict) - answer_dict = json.loads(answer) - assert answer_dict['FeatureEnabledCollectionByMap'] is True - assert answer_dict['FeatureEnabledCollectionByToken'] is False - assert answer_dict['PermittedCreateCollectionByMap'] is True - assert answer_dict['PermittedCreateCollectionByToken'] is True + # path with valid API and hook but not enabled "token" + self.configure({"sharing": { + "collection_by_map": "True", + "collection_by_token": "False"} + }) + sharetype = "token" + for action in sharing.API_HOOKS_V1: + path = "/.sharing/v1/" + sharetype + "/" + action + logging.info("\n*** check valid API URI hook (but not enabled): %r", path) + _, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw") - logging.info("\n*** check API hook: info/map") - json_dict = {} - _, headers, answer = self._sharing_api_json("map", "info", check=200, login="owner:ownerpw", json_dict=json_dict) - answer_dict = json.loads(answer) - assert answer_dict['FeatureEnabledCollectionByMap'] is True - assert 'FeatureEnabledCollectionByToken' not in answer_dict - assert 'PermittedCreateCollectionByToken' not in answer_dict + # check info hook + logging.info("\n*** check API hook: info/all") + json_dict = {} + _, headers, answer = self._sharing_api_json("all", "info", check=200, login="owner:ownerpw", json_dict=json_dict) + answer_dict = json.loads(answer) + assert answer_dict['FeatureEnabledCollectionByMap'] is True + assert answer_dict['FeatureEnabledCollectionByToken'] is False + assert answer_dict['PermittedCreateCollectionByMap'] is True + assert answer_dict['PermittedCreateCollectionByToken'] is True - logging.info("\n*** check API hook: info/token -> 404 (not enabled)") - json_dict = {} - _, headers, answer = self._sharing_api_json("token", "info", check=404, login="owner:ownerpw", json_dict=json_dict) + logging.info("\n*** check API hook: info/map") + json_dict = {} + _, headers, answer = self._sharing_api_json("map", "info", check=200, login="owner:ownerpw", json_dict=json_dict) + answer_dict = json.loads(answer) + assert answer_dict['FeatureEnabledCollectionByMap'] is True + assert 'FeatureEnabledCollectionByToken' not in answer_dict + assert 'PermittedCreateCollectionByToken' not in answer_dict - # path with valid API and hook and all enabled - self.configure({"sharing": { - "collection_by_map": "True", - "collection_by_token": "True"} - }) - for sharetype in sharing.SHARE_TYPES: - path = "/.sharing/v1/" + sharetype + "/" + action - # invalid API - _, headers, _ = self.request("POST", path + "NA", check=404, login="owner:ownerpw") - # valid API - _, headers, _ = self.request("POST", path, check=400, login="owner:ownerpw") + logging.info("\n*** check API hook: info/token -> 404 (not enabled)") + json_dict = {} + _, headers, answer = self._sharing_api_json("token", "info", check=404, login="owner:ownerpw", json_dict=json_dict) - logging.info("\n*** check API hook: info/token -> 200") - json_dict = {} - _, headers, answer = self._sharing_api_json("token", "info", check=200, login="owner:ownerpw", json_dict=json_dict) - answer_dict = json.loads(answer) - assert answer_dict['FeatureEnabledCollectionByToken'] is True - assert 'FeatureEnabledCollectionByMap' not in answer_dict - assert 'PermittedCreateCollectionByMap' not in answer_dict + # path with valid API and hook and all enabled + self.configure({"sharing": { + "collection_by_map": "True", + "collection_by_token": "True"} + }) + for sharetype in sharing.SHARE_TYPES: + path = "/.sharing/v1/" + sharetype + "/" + action + # invalid API + _, headers, _ = self.request("POST", path + "NA", check=404, login="owner:ownerpw") + # valid API + _, headers, _ = self.request("POST", path, check=400, login="owner:ownerpw") + + logging.info("\n*** check API hook: info/token -> 200") + json_dict = {} + _, headers, answer = self._sharing_api_json("token", "info", check=200, login="owner:ownerpw", json_dict=json_dict) + answer_dict = json.loads(answer) + assert answer_dict['FeatureEnabledCollectionByToken'] is True + assert 'FeatureEnabledCollectionByMap' not in answer_dict + assert 'PermittedCreateCollectionByMap' not in answer_dict def test_sharing_api_list_with_auth(self) -> None: """POST/list with authentication.""" @@ -196,9 +234,7 @@ class TestSharingApiSanity(BaseTest): form_array: Sequence[str] json_dict: dict - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -286,9 +322,7 @@ class TestSharingApiSanity(BaseTest): form_array: Sequence[str] json_dict: dict - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -361,7 +395,7 @@ class TestSharingApiSanity(BaseTest): _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array, accept="text/csv") assert "Status=success" not in answer assert "Lines=2" not in answer - assert ",".join(sharing.DB_FIELDS_V1) in answer + assert ";".join(sharing.DB_FIELDS_V1) in answer assert "/owner/collection1/" in answer assert "/owner/collection2/" in answer @@ -408,7 +442,7 @@ class TestSharingApiSanity(BaseTest): _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) assert "Status=success" in answer assert "Lines=1" in answer - assert "True,True,True,True" in answer + assert "True;True;True;True" in answer logging.info("\n*** hide token#2 (form->text)") form_array = [] @@ -422,7 +456,7 @@ class TestSharingApiSanity(BaseTest): _, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array) assert "Status=success" in answer assert "Lines=1" in answer - assert "True,True,True,True" in answer + assert "True;True;True;True" in answer logging.info("\n*** unhide token#2 (json->json)") json_dict = {} @@ -484,9 +518,7 @@ class TestSharingApiSanity(BaseTest): path = path_base + "/event1.ics" self.put(path, event, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -592,9 +624,7 @@ class TestSharingApiSanity(BaseTest): json_dict: dict - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -645,9 +675,7 @@ class TestSharingApiSanity(BaseTest): event = get_file_content(file_item2) self.put(path_mapped_item2, event, check=201, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -834,9 +862,7 @@ class TestSharingApiSanity(BaseTest): path = path_mapped2 + "/event1.ics" self.put(path, event, login="%s:%s" % ("owner2", "owner2pw")) - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -923,9 +949,7 @@ class TestSharingApiSanity(BaseTest): path = path_mapped + "/event1.ics" self.put(path, event, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1117,9 +1141,7 @@ class TestSharingApiSanity(BaseTest): event = get_file_content("event1.ics") self.put(path_mapped_item, event, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1228,9 +1250,7 @@ class TestSharingApiSanity(BaseTest): event = get_file_content("event2.ics") self.put(path_user_item, event, login="user:userpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1364,9 +1384,7 @@ class TestSharingApiSanity(BaseTest): path = os.path.join(path_mapped, "event1.ics") self.put(path, event, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1439,7 +1457,7 @@ class TestSharingApiSanity(BaseTest): element = prop.find(xmlutils.make_clark("D:href")) assert element is not None and element.text == "/user/" - def test_sharing_api_map_proppatch(self) -> None: + def test_sharing_api_map_proppatch_acl(self) -> None: """share-by-map API usage tests related to report.""" self.configure({"auth": {"type": "htpasswd", "htpasswd_filename": self.htpasswd_file_path, @@ -1468,9 +1486,7 @@ class TestSharingApiSanity(BaseTest): path = os.path.join(path_mapped, "event1.ics") self.put(path, event, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1647,9 +1663,7 @@ class TestSharingApiSanity(BaseTest): event = get_file_content("event3.ics") self.put(os.path.join(path_user, "event3.ics"), event, login="user:userpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1837,9 +1851,7 @@ class TestSharingApiSanity(BaseTest): event = get_file_content("event1.ics") self.put(os.path.join(path_mapped1, "event1.ics"), event, login="owner:ownerpw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -1959,9 +1971,7 @@ class TestSharingApiSanity(BaseTest): self.mkcalendar(path_user1, login="user1:user1pw") self.mkcalendar(path_user2, login="user2:user2pw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -2107,9 +2117,7 @@ class TestSharingApiSanity(BaseTest): logging.info("\n*** mkcalendar user2 -> conflict") self.mkcalendar(path_user2, login="user2:user2pw", check=409) - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -2179,9 +2187,7 @@ class TestSharingApiSanity(BaseTest): logging.info("\n*** prepare") self.mkcalendar(path_owner1, login="owner1:owner1pw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -2272,9 +2278,7 @@ permissions: RrWw""") self.mkcalendar(path_owner1_M, login="owner1:owner1pw") self.mkcalendar(path_owner1_m, login="owner1:owner1pw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -2394,9 +2398,7 @@ permissions: RrWw""") logging.info("\n*** prepare") self.mkcalendar(path_owner1, login="owner1:owner1pw") - for db_type in sharing.INTERNAL_TYPES: - if db_type == "none": - continue + 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}}) @@ -2492,3 +2494,199 @@ permissions: RrWw""") assert answer_dict['Status'] == "success" assert answer_dict['Lines'] == 1 assert answer_dict['Content'][0]['Permissions'] == "RrWw" + + def test_sharing_api_map_propfind_overlay(self) -> None: + """share-by-map API usage tests related to proppatch.""" + 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, + "collection_by_map": "True", + "collection_by_token": "True"}, + "logging": {"request_header_on_debug": "False", + "response_content_on_debug": "True", + "request_content_on_debug": "True"}, + "rights": {"type": "owner_only"}}) + + form_array: Sequence[str] + json_dict: dict + + path_mapped = "/owner/calendarPFO.ics/" + path_shared_r = "/user/calendarPFO-shared-by-owner-r.ics/" + + logging.info("\n*** prepare and test access") + self.mkcalendar(path_mapped, login="owner:ownerpw") + + 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}}) + + # check PROPFIND as owner + logging.info("\n*** PROPFIND collection owner -> ok") + _, responses = self.propfind(path_mapped, """\ + + + + + +""", login="owner:ownerpw") + logging.info("response: %r", responses) + response = responses[path_mapped] + assert not isinstance(response, int) and len(response) == 1 + status, prop = response["D:current-user-principal"] + assert status == 200 and len(prop) == 1 + element = prop.find(xmlutils.make_clark("D:href")) + assert element is not None and element.text == "/owner/" + + # execute PROPPATCH as owner + logging.info("\n*** PROPPATCH collection owner -> ok") + _, responses = self.proppatch(path_mapped, """\ + + + + + #AAAAAA + ICAL-OWNER + + +""", login="owner:ownerpw") + logging.info("response: %r", responses) + response = responses[path_mapped] + assert not isinstance(response, int) and len(response) == 2 + status, prop = response["ICAL:calendar-color"] + assert status == 200 and not prop.text + status, prop = response["C:calendar-description"] + assert status == 200 and not prop.text + + # verify PROPPATCH by owner + logging.info("\n*** PROPFIND collection owner -> ok") + propfind_calendar_color = get_file_content("propfind_multiple.xml") + _, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw") + logging.info("response: %r", responses) + response = responses[path_mapped] + assert not isinstance(response, int) + status, prop = response["C:calendar-description"] + logging.debug("calendar-description: %r", prop.text) + assert status == 200 and prop.text == "ICAL-OWNER" + status, prop = response["ICAL:calendar-color"] + logging.debug("calendar-color: %r", prop.text) + assert status == 200 and prop.text == "#AAAAAA" + + # create map + 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" + 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" + + # enable map by user + logging.info("\n*** enable map by user") + 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) + + # verify PROPPATCH as user + logging.info("\n*** PROPFIND collection user -> ok") + propfind_calendar_color = get_file_content("propfind_multiple.xml") + _, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw") + logging.info("response: %r", responses) + response = responses[path_mapped] + assert not isinstance(response, int) + status, prop = response["C:calendar-description"] + logging.debug("calendar-description: %r", prop.text) + assert status == 200 and prop.text == "ICAL-OWNER" + status, prop = response["ICAL:calendar-color"] + logging.debug("calendar-color: %r", prop.text) + assert status == 200 and prop.text == "#AAAAAA" + + # 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) + + logging.info("\n*** list (json->json)") + json_dict['PathOrToken'] = path_shared_r + _, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner:ownerpw", json_dict=json_dict) + answer_dict = json.loads(answer) + assert answer_dict['Status'] == "success" + assert answer_dict['Lines'] == 1 + + logging.info("\n*** list (json->csv)") + json_dict['PathOrToken'] = path_shared_r + _, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/csv") + + logging.info("\n*** list (json->txt)") + json_dict['PathOrToken'] = path_shared_r + _, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain") + + # verify overlay as user + logging.info("\n*** PROPFIND collection user (overlay) -> ok") + propfind_calendar_color = get_file_content("propfind_multiple.xml") + _, responses = self.propfind(path_shared_r, propfind_calendar_color, login="user:userpw") + logging.info("response: %r", responses) + response = responses[path_shared_r] + assert not isinstance(response, int) + status, prop = response["C:calendar-description"] + logging.debug("calendar-description: %r", prop.text) + assert status == 200 and prop.text == "ICAL-USER" + status, prop = response["ICAL:calendar-color"] + logging.debug("calendar-color: %r", prop.text) + assert status == 200 and prop.text == "#BBBBBB" + + # verify overlay not visible by owner + logging.info("\n*** PROPFIND collection owner -> ok") + propfind_calendar_color = get_file_content("propfind_multiple.xml") + _, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw") + logging.info("response: %r", responses) + response = responses[path_mapped] + assert not isinstance(response, int) + status, prop = response["C:calendar-description"] + logging.debug("calendar-description: %r", prop.text) + assert status == 200 and prop.text == "ICAL-OWNER" + status, prop = response["ICAL:calendar-color"] + logging.debug("calendar-color: %r", prop.text) + assert status == 200 and prop.text == "#AAAAAA" + + # update map by user + logging.info("\n*** update map by user (form)") + form_array = ["User=" + "user"] + 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'") + _, headers, answer = self._sharing_api_form("map", "update", check=200, login="user:userpw", form_array=form_array) + assert "Status=success" in answer + + # verify overlay as user + logging.info("\n*** PROPFIND collection user (overlay) -> ok") + propfind_calendar_color = get_file_content("propfind_multiple.xml") + _, responses = self.propfind(path_shared_r, propfind_calendar_color, login="user:userpw") + logging.info("response: %r", responses) + response = responses[path_shared_r] + assert not isinstance(response, int) + status, prop = response["C:calendar-description"] + logging.debug("calendar-description: %r", prop.text) + assert status == 200 and prop.text == "ICAL-USER-NEW" + status, prop = response["ICAL:calendar-color"] + logging.debug("calendar-color: %r", prop.text) + assert status == 200 and prop.text == "#CCCCCC" + + # update map by user + logging.info("\n*** update map by user (form)") + form_array = ["User=" + "user"] + form_array.append("PathOrToken=" + path_shared_r) + form_array.append("Properties=BUGGYENTRY=BUGGYVALUE") + _, headers, answer = self._sharing_api_form("map", "update", check=400, login="user:userpw", form_array=form_array) diff --git a/radicale/utils.py b/radicale/utils.py index 8d7ba914..9e59dc7b 100644 --- a/radicale/utils.py +++ b/radicale/utils.py @@ -385,7 +385,7 @@ def limit_str(content: str, limit: int) -> str: return content -def textwrap_str(content: str, limit: int = 2000) -> str: +def textwrap_str(content: str, limit: int = 3000) -> str: # TODO: add support for config option and prefix return textwrap.indent(limit_str(content, limit), " ", lambda line: True)