diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 337bb135..dff4fd46 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -85,6 +85,24 @@ jobs: - name: Test with newest Python on latest Ubuntu using VFAT run: tox -c pyproject.toml -e py_filesystem_vfat + test-ubuntu-python-newest-with-vfat-utf8: + name: Test VFAT UTF-8 Python:newest Ubuntu:latest + needs: [lint, test-ubuntu-python-newest, test-ubuntu-python-newest-with-vfat] + strategy: + matrix: + os: [ubuntu-latest] + python-version: ['3.14'] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v5 + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + - name: Install Test dependencies + run: pip install tox + - name: Test with newest Python on latest Ubuntu using VFAT UTF-8 + run: tox -c pyproject.toml -e py_filesystem_vfat_utf8 + test-ubuntu-python-newest-with-ntfs: name: Test NTFS Python:newest Ubuntu:latest needs: [lint, test-ubuntu-python-newest] @@ -163,7 +181,7 @@ jobs: test-otheros-python-newest: name: Test MacOS/Windows:latest Python:newest - needs: [lint, test-ubuntu-python-newest, test-ubuntu-python-newest-with-ntfs, test-ubuntu-python-newest-with-vfat] + needs: [lint, test-ubuntu-python-newest, test-ubuntu-python-newest-with-ntfs, test-ubuntu-python-newest-with-vfat, test-ubuntu-python-newest-with-vfat-utf8] strategy: matrix: os: [macos-latest, windows-latest] @@ -310,7 +328,7 @@ jobs: - uses: actions/checkout@v5 - uses: actions/setup-python@v6 with: - python-version: '3.12' + python-version: '3.14' - name: Install uv run: pip install uv - name: Install Playwright Browsers diff --git a/CHANGELOG.md b/CHANGELOG.md index edd4c501..995283c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,11 @@ * Improve: application will stop on startup if TEMP is provided but not existing or not writable * Extension: tox with new optional test cases to test with LinuxOS vfat, hfsplus, ntfs filesystems * Adjust: respond with 500 in case principal collection cannot be created (e.g. filesystem issues) +* Improve: sharing supports now unicode +* Add: [server] new options validate_user_value/validate_path_value for ability to block unwanted values (autoenable "strict" on non-unicode filesystem) +* Adjust: several log levels incl. final result depending on status code +* Fix: sharing/csv: quote handling and honor stock encoding +* Extension: sharing: add ICAL:calendar-order to property overlay whitelist ## 3.7.1 * Fix: share address book collection as birthday calendar not working on non-DEBUG level diff --git a/DOCUMENTATION.md b/DOCUMENTATION.md index 6f1a570b..fada9827 100644 --- a/DOCUMENTATION.md +++ b/DOCUMENTATION.md @@ -1617,6 +1617,35 @@ Strict preconditions check on PUT in case item already exists [RFC6352#9.2](http Default: `False` +##### validate_user_value + +_(>= 3.7.2)_ + +Validate user value content + +Available types are: +* `none` +* `minimal` (control and some special chars) +* `unicode-letter` (unicode letters) +* `no-unicode` (no unicode) +* `strict` (reduced ASCII set) + +Default: `minimum` + +##### validate_path_type + +_(>= 3.7.2)_ + +Validate path value content + +* `none` +* `minimal` (control and some special chars) +* `unicode-letter` (unicode letters) +* `no-unicode` (no unicode) +* `strict` (reduced ASCII set) + +Default: `minimum` + ##### hook Command that is run after changes to storage. See the diff --git a/config b/config index d899718d..77fe4805 100644 --- a/config +++ b/config @@ -58,6 +58,14 @@ # script name to strip from URI if called by reverse proxy #script_name = (default taken from HTTP_X_SCRIPT_NAME or SCRIPT_NAME) +# validate user type +# Value: none|minimal|unicode-letter|no-unicode|strict +#validate_user_value = minimal + +# validate path value +# Value: none|minimal|unicode-letter|no-unicode|strict +#validate_path_value = minimal + [encoding] @@ -424,6 +432,7 @@ # This may become the default in future versions, override if you need a different CSP. Content-Security-Policy = default-src 'self'; object-src 'none' + [hook] # Hook types diff --git a/pyproject.toml b/pyproject.toml index fddb1cb9..e9865805 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -101,7 +101,7 @@ commands_pre = [ # unconditionally create mount point ["mkdir", "-p", "/tmp/vfat"], # mount image - ["sudo", "/usr/bin/mount", "-o", "loop,umask=000", "/tmp/vfat.img", "/tmp/vfat"], + ["sudo", "/usr/bin/mount", "-o", "loop,umask=000,utf8=no", "/tmp/vfat.img", "/tmp/vfat"], ] setenv = { TEMP = "/tmp/vfat" } commands = [["pytest", "-r", "s", "."]] @@ -114,6 +114,33 @@ commands_post = [ ["rm", "/tmp/vfat.img"] ] +[tool.tox.env.py_filesystem_vfat_utf8] +allowlist_externals = [ "sudo", "dd", "mkfs.vfat", "mkdir", "rm", "rmdir", "chmod" ] +extras = ["test"] +deps = [ + "pytest" +] +commands_pre = [ + # create 64 MByte disk image + ["dd", "if=/dev/zero", "of=/tmp/vfat-utf8.img", "bs=1M", "count=64"], + # create file system + ["mkfs.vfat", "/tmp/vfat-utf8.img", "-n", "VFAT"], + # unconditionally create mount point + ["mkdir", "-p", "/tmp/vfat-utf8"], + # mount image + ["sudo", "/usr/bin/mount", "-o", "loop,umask=000,utf8", "/tmp/vfat-utf8.img", "/tmp/vfat-utf8"], +] +setenv = { TEMP = "/tmp/vfat-utf8" } +commands = [["pytest", "-r", "s", "."]] +commands_post = [ + # umount image + ["sudo", "/usr/bin/umount", "-d", "/tmp/vfat-utf8"], + # remove mount point + ["rmdir", "/tmp/vfat-utf8"], + # remove image + ["rm", "/tmp/vfat-utf8.img"] +] + [tool.tox.env.py_filesystem_hfsplus] # Fedora allowlist_externals = [ "sudo", "dd", "mkfs.hfsplus", "mkdir", "rm", "rmdir", "chmod" ] diff --git a/radicale/app/__init__.py b/radicale/app/__init__.py index 9956e023..46814e44 100644 --- a/radicale/app/__init__.py +++ b/radicale/app/__init__.py @@ -42,6 +42,7 @@ from http import client from typing import Iterable, List, Mapping, Tuple, Union from radicale import config, httputils, log, pathutils, types, utils +from radicale.app import base as app_base from radicale.app.base import ApplicationBase from radicale.app.delete import ApplicationPartDelete from radicale.app.get import ApplicationPartGet @@ -86,6 +87,8 @@ class Application(ApplicationPartDelete, ApplicationPartHead, _profiling_per_request: bool = False _profiling_per_request_method: bool = False _limit_content: int + _validate_user_value: str + _validate_path_value: str profiler_per_request_method: dict[str, cProfile.Profile] = {} profiler_per_request_method_counter: dict[str, int] = {} profiler_per_request_method_starttime: datetime.datetime @@ -158,6 +161,19 @@ class Application(ApplicationPartDelete, ApplicationPartHead, self._extra_headers[key] = configuration.get("headers", key) self._strict_preconditions = configuration.get("storage", "strict_preconditions") logger.info("strict preconditions check: %s", self._strict_preconditions) + # Format checks + self._validate_user_value = configuration.get("server", "validate_user_value") + self._validate_path_value = configuration.get("server", "validate_path_value") + if not self._storage._supports_unicode: + if self._validate_user_value not in ["strict", "no-unicode"]: + self._validate_user_value = "no-unicode" + if self._validate_path_value not in ["strict", "no-unicode"]: + self._validate_path_value = "no-unicode" + logger.notice("validate user value: %r (enforced by limited support of collection storage)", self._validate_user_value) + logger.notice("validate path value: %r (enforced by limited support of collection storage)", self._validate_path_value) + else: + logger.info("validate user value: %r", self._validate_user_value) + logger.info("validate path value: %r", self._validate_path_value) # Profiling options self._profiling = configuration.get("logging", "profiling") self._profiling_per_request_min_duration = configuration.get("logging", "profiling_per_request_min_duration") @@ -338,15 +354,23 @@ class Application(ApplicationPartDelete, ApplicationPartHead, else: flags_text = "" if answer is not None: - logger.info("%s response status for %r%s in %.3f seconds %s %s bytes%s: %s", + message = "%s response status for %r%s in %.3f seconds %s %s bytes%s: %s" % ( request_method, unsafe_path, depthinfo, time_delta_seconds, content_encoding, str(len(answer)), flags_text, status_text) else: - logger.info("%s response status for %r%s in %.3f seconds: %s", + message = "%s response status for %r%s in %.3f seconds: %s" % ( request_method, unsafe_path, depthinfo, time_delta_seconds, status_text) + logger_method = logger.info # default + if status == 401 or status == 404 or status == 412 or status == 409 or request_method in ["PROPPATCH", "MKCALENDAR", "MKCOL"]: + logger_method = logger.notice + elif status >= 500 and status < 500: + logger_method = logger.error + elif status >= 500: + logger_method = logger.critical + logger_method(message) # Profiling end if self._profiling_per_request: @@ -459,6 +483,9 @@ class Application(ApplicationPartDelete, ApplicationPartHead, logger.warning("Called by reverse proxy, cannot remove base prefix %r from path: %r as not matching (may cause authentication issues using internal WebUI)", base_prefix, path) else: logger.debug("Called by reverse proxy, cannot remove base prefix %r from path: %r as not matching", base_prefix, path) + if not app_base._check_path_format(self._storage, path, self._validate_path_value): + logger.error("request contains invalid path: %r (not compliant to %r)", path, self._validate_path_value) + return response(*httputils.BAD_REQUEST) # Get function corresponding to method function = getattr(self, "do_%s" % request_method, None) @@ -489,7 +516,11 @@ class Application(ApplicationPartDelete, ApplicationPartHead, self.configuration, environ, base64.b64decode( authorization.encode("ascii"))).split(":", 1) - (user, info) = self._auth.login(login, password, context) or ("", "") if login else ("", "") + if login and not app_base._check_user_format(self._storage, login, self._validate_user_value): + info = "not compliant to %r" % self._validate_user_value + user = "" + else: + (user, info) = self._auth.login(login, password, context) or ("", "") if login else ("", "") if self.configuration.get("auth", "type") == "ldap": try: logger.debug("Groups received from LDAP: %r", ",".join(self._auth._ldap_groups)) @@ -600,9 +631,9 @@ class Application(ApplicationPartDelete, ApplicationPartHead, if (status, headers, answer, xml_request) == httputils.NOT_ALLOWED: if path.startswith("/.token"): - logger.info("Access to %r denied", path) + logger.notice("Access to %r denied", path) else: - logger.info("Access to %r denied for %s", path, repr(user) if user else "anonymous user") + logger.notice("Access to %r denied for %s", path, repr(user) if user else "anonymous user") else: status, headers, answer, xml_request = httputils.NOT_ALLOWED diff --git a/radicale/app/base.py b/radicale/app/base.py index cff58ec1..63993a9a 100644 --- a/radicale/app/base.py +++ b/radicale/app/base.py @@ -17,7 +17,9 @@ import io import logging +import re import sys +import unicodedata import xml.etree.ElementTree as ET from typing import Optional, Union @@ -30,6 +32,82 @@ from radicale.rights import intersect import defusedxml.ElementTree as DefusedET # isort:skip sys.modules["xml.etree"].ElementTree = ET # type:ignore[attr-defined] +USER_PATTERN_STRICT: str = "a-zA-Z0-9@\\.\\-_" +PATH_PATTERN_STRICT: str = USER_PATTERN_STRICT + "\\/~" # / as separator + +USER_PATTERN_STRICT_RE: str = "^[" + USER_PATTERN_STRICT + "]+$" +PATH_PATTERN_STRICT_RE: str = "^[" + PATH_PATTERN_STRICT + "]+$" + +USER_BLACKLIST_MINIMAL: list = [":", "'", '"', '*', '?'] +PATH_BLACKLIST_MINIMAL: list = USER_BLACKLIST_MINIMAL + +USER_WHITELIST_UNICODE: list = ["-", ".", "@", "_"] # from USER_PATTERN_STRICT +PATH_WHITELIST_UNICODE: list = ["-", ".", "@", "_", "/", "~"] # from PATH_PATTERN_STRICT + + +def _check_format(self: storage.BaseStorage, + string: str, + blacklist_minimal: list[str], + whitelist_unicode: list[str], + validation_type: str, + ) -> bool: + check_minimal = (validation_type == "minimal") + check_unicode_letter = (validation_type == "unicode-letter") + check_no_unicode = (validation_type == "no-unicode") + logger.trace("_check_format investigate %r (validation_type=%r check_minimal=%s check_unicode_letter=%s check_no_unicode=%s)", string, validation_type, check_minimal, check_unicode_letter, check_no_unicode) + if not self._supports_trailing_whitespace and string.endswith(' '): + return False + for c in string: + if c <= chr(31) or (c >= chr(127) and c <= chr(159)): + # ASCII: control char + return False + if unicodedata.category(c)[0] == "C": + # https://unicodeplus.com/category + # Unicode: control + return False + if check_minimal or not self._supports_problematic_chars: + if c in blacklist_minimal: + logger.trace("_check_format found %r", c) + return False + if check_unicode_letter: + if c not in whitelist_unicode: + if unicodedata.category(c)[0] != "L": + return False + if check_no_unicode: + if ord(c) > 255: + return False + return True + + +def _check_user_format(self: storage.BaseStorage, + user: str, + validation_type: str + ) -> bool: + if validation_type == "strict": + return (re.search(USER_PATTERN_STRICT_RE, user) is not None) + else: + return _check_format(self, + user, + USER_BLACKLIST_MINIMAL, + USER_WHITELIST_UNICODE, + validation_type, + ) + + +def _check_path_format(self: storage.BaseStorage, + path: str, + validation_type: str + ) -> bool: + if validation_type == "strict": + return (re.search(PATH_PATTERN_STRICT_RE, path) is not None) + else: + return _check_format(self, + path, + PATH_BLACKLIST_MINIMAL, + PATH_WHITELIST_UNICODE, + validation_type, + ) + class ApplicationBase: @@ -44,6 +122,8 @@ class ApplicationBase: _permit_delete_collection: bool _permit_overwrite_collection: bool _strict_preconditions: bool + _validate_user_value: str + _validate_path_format: str _hook: hook.BaseHook def __init__(self, configuration: config.Configuration) -> None: @@ -58,6 +138,8 @@ class ApplicationBase: self._response_content_on_debug = configuration.get("logging", "response_content_on_debug") self._request_content_on_debug = configuration.get("logging", "request_content_on_debug") self._limit_content = configuration.get("logging", "limit_content") + self._validate_user_value = configuration.get("server", "validate_user_value") + self._validate_path_value = configuration.get("server", "validate_path_value") self._hook = hook.load(configuration) def _read_xml_request_body(self, environ: types.WSGIEnviron diff --git a/radicale/app/move.py b/radicale/app/move.py index 9af6e378..f16368f4 100644 --- a/radicale/app/move.py +++ b/radicale/app/move.py @@ -25,6 +25,7 @@ from http import client from urllib.parse import unquote, urlparse from radicale import httputils, pathutils, storage, types +from radicale.app import base as app_base from radicale.app.base import Access, ApplicationBase from radicale.log import logger @@ -82,6 +83,9 @@ class ApplicationPartMove(ApplicationBase): if not access.check("w"): return httputils.NOT_ALLOWED to_path = pathutils.sanitize_path(to_url.path) + if not app_base._check_path_format(self._storage, to_path, self._validate_path_value): + logger.warning("request contains invalid path: %r (not compliant to %r)", to_path, self._validate_path_value) + return httputils.BAD_REQUEST if not (to_path + "/").startswith(base_prefix + "/"): logger.warning("Destination %r from MOVE request on %r doesn't " "start with base prefix", to_path, path) diff --git a/radicale/app/proppatch.py b/radicale/app/proppatch.py index 0f60dc97..2fd48671 100644 --- a/radicale/app/proppatch.py +++ b/radicale/app/proppatch.py @@ -127,13 +127,13 @@ class ApplicationPartProppatch(ApplicationBase): (not self._sharing.permit_properties_overlay and "P" not in raw_permissions and "P" not in share['Permissions'])): logger.info("PROPPATCH request on shared %r: write-access", path_orig) if permissions_filter is not None and "e" in permissions_filter: - logger.info("PROPPATCH request on shared %r: write-access, overlay enforced, but disabled by share permission 'e'", path_orig) + logger.notice("PROPPATCH request on shared %r: write-access, overlay enforced, but disabled by share permission 'e'", path_orig) elif "e" in raw_permissions: - logger.info("PROPPATCH request on shared %r: write-access, overlay enforced, but disabled by rights permission 'e'", path_orig) + logger.notice("PROPPATCH request on shared %r: write-access, overlay enforced, but disabled by rights permission 'e'", path_orig) else: share_overlay = True else: - logger.info("PROPPATCH request on shared %r: no write-access", path_orig) + logger.notice("PROPPATCH request on shared %r: no write-access", path_orig) return httputils.NOT_ALLOWED else: return httputils.NOT_ALLOWED @@ -144,9 +144,9 @@ class ApplicationPartProppatch(ApplicationBase): logger.trace("PROPPATCH/xml_proppatch: write-access/sharing: %r", path_orig) if self._sharing.enforce_properties_overlay: if permissions_filter is not None and "e" in permissions_filter: - logger.info("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by share permission 'e'", path_orig) + logger.notice("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by share permission 'e'", path_orig) elif "e" in raw_permissions: - logger.info("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by rights permission 'e'", path_orig) + logger.notice("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by rights permission 'e'", path_orig) else: share_overlay = True else: diff --git a/radicale/app/put.py b/radicale/app/put.py index 5f916fe0..08f1bb0a 100644 --- a/radicale/app/put.py +++ b/radicale/app/put.py @@ -365,7 +365,7 @@ class ApplicationPartPut(ApplicationBase): errno_match = re.search("\\[Errno ([0-9]+)\\]", str(e)) if errno_match: logger.error( - "Failed PUT request on %r (upload): %s", path, e, exc_info=True) + "Failed PUT request on %r (upload): %s", path, e, exc_info=False) errno_e = int(errno_match.group(1)) if errno_e == errno.ENOSPC: return httputils.INSUFFICIENT_STORAGE diff --git a/radicale/config.py b/radicale/config.py index 581d4939..a592f2d0 100644 --- a/radicale/config.py +++ b/radicale/config.py @@ -48,6 +48,8 @@ DEFAULT_CONFIG_PATH: str = os.pathsep.join([ PROFILING: Sequence[str] = ("per_request", "per_request_method", "none") +VALIDATE_TYPES: Sequence[str] = ("none", "minimal", "unicode-letter", "unicode-none", "strict") + def positive_int(value: Any) -> int: value = int(value) @@ -84,6 +86,12 @@ def logging_level(value: Any) -> str: return value +def validate_types(value: Any) -> str: + if value not in VALIDATE_TYPES: + raise ValueError("unsupported validation type: %r" % value) + return value + + def profiling(value: Any) -> str: if value not in PROFILING: raise ValueError("unsupported profiling: %r" % value) @@ -221,6 +229,14 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([ "value": "", "help": "script name to strip from URI if called by reverse proxy (default taken from HTTP_X_SCRIPT_NAME or SCRIPT_NAME)", "type": str}), + ("validate_user_value", { + "value": "minimal", + "help": "validate user value (" + "|".join(VALIDATE_TYPES) + ")", + "type": validate_types}), + ("validate_path_value", { + "value": "minimal", + "help": "validate path value (" + "|".join(VALIDATE_TYPES) + ")", + "type": validate_types}), ("_internal_server", { "value": "False", "help": "the internal server is used", diff --git a/radicale/pathutils.py b/radicale/pathutils.py index 31d6d875..65a80e18 100644 --- a/radicale/pathutils.py +++ b/radicale/pathutils.py @@ -289,22 +289,14 @@ def path_to_filesystem(root: str, sane_path: str, path_is_collision_free: bool = raise UnsafePathError(part) safe_path_parent = safe_path safe_path = os.path.join(safe_path, part) - # Check for conflicting files (e.g. case-insensitive file systems - # or short names on Windows file systems) if not path_is_collision_free: - if sys.platform == "win32" and False: # temporary for testing - # logger.trace("path_to_filesystem check (win32): %r", part) - # if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)) and not os.path.islink(safe_path): - if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)): - raise CollidingPathError(part) - else: - # logger.trace("path_to_filesystem check (!win32): %r", part) - if os.path.lexists(safe_path): - with os.scandir(safe_path_parent) as entries: - if part not in (e.name for e in entries): - raise CollidingPathError(part) + # Check for conflicting files (e.g. case-insensitive file systems + # or short names on Windows file systems) + if os.path.lexists(safe_path): + with os.scandir(safe_path_parent) as entries: + if part not in (e.name for e in entries): + raise CollidingPathError(part) else: - # logger.trace("path_to_filesystem check (skipped): %r", part) pass return safe_path diff --git a/radicale/sharing/__init__.py b/radicale/sharing/__init__.py index 41b8067b..ba2c5f6b 100644 --- a/radicale/sharing/__init__.py +++ b/radicale/sharing/__init__.py @@ -121,11 +121,7 @@ API_TYPES_V1: dict[str, type] = { TOKEN_PATTERN_V1: str = "v1/[a-zA-Z0-9_\\-]{44}" -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", "D:displayname") +OVERLAY_PROPERTIES_WHITELIST: Sequence[str] = ("C:calendar-description", "ICAL:calendar-color", "CR:addressbook-description", "INF:addressbook-color", "D:displayname", "ICAL:calendar-order") CONVERSIONS_WHITELIST: Sequence[str] = ("bday", "none") @@ -141,6 +137,7 @@ class BaseSharing: _rights: rights.BaseRights _auth_delay: float _enabled: bool = False + _encoding: str default_permissions_create_token: str default_permissions_create_map: str sharing_db_type: str @@ -157,6 +154,9 @@ class BaseSharing: self._rights = rights.load(configuration) self._storage = storage.load(configuration) self._auth_delay = configuration.get("auth", "delay") + self._encoding = configuration.get("encoding", "stock") + self._validate_user_value = configuration.get("server", "validate_user_value") + self._validate_path_value = configuration.get("server", "validate_path_value") # Sharing self.sharing_collection_by_map = configuration.get("sharing", "collection_by_map") self.sharing_collection_by_token = configuration.get("sharing", "collection_by_token") @@ -508,6 +508,7 @@ class BaseSharing: # *** POST API *** def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse: # Late import to avoid circular dependency in config + from radicale.app import base as app_base from radicale.app.base import Access """POST request. @@ -724,20 +725,20 @@ class BaseSharing: logger.warning(api_info + ": unsupported " + key) return httputils.bad_request("Invalid value for PathOrToken") else: - if not re.search('^' + PATH_PATTERN + '$', request_data[key]): - logger.warning(api_info + ": unsupported " + key) + if not app_base._check_path_format(self._storage, request_data[key], self._validate_path_value): + logger.warning("%s: invalid %r: %r (not compliant to %r)", api_info, key, request_data[key], self._validate_path_value) return httputils.bad_request("Invalid value for PathOrToken") if not request_data[key].endswith("/"): return httputils.bad_request("PathOrToken not ending with /") elif key == "PathMapped": - if not re.search('^' + PATH_PATTERN + '$', request_data[key]): - logger.warning(api_info + ": unsupported " + key) + if not app_base._check_path_format(self._storage, request_data[key], self._validate_path_value): + logger.warning("%s: invalid %r: %r (not compliant to %r)", api_info, key, request_data[key], self._validate_path_value) return httputils.bad_request("Invalid value for PathMapped") elif not request_data[key].endswith("/"): return httputils.bad_request("PathMapped not ending with /") elif key == "User": - if not re.search('^' + USER_PATTERN + '$', request_data[key]): - logger.warning(api_info + ": unsupported " + key) + if not app_base._check_user_format(self._storage, request_data[key], self._validate_user_value): + logger.warning("%s: invalid %r: %r (not compliant to %r)", api_info, key, request_data[key], self._validate_user_value) return httputils.bad_request("Invalid value for User") # check for optional parameters @@ -852,6 +853,12 @@ class BaseSharing: logger.warning(api_info + ": missing PathMapped") return httputils.bad_request("Missing PathMapped") + # check access Permissions + access = Access(self._rights, user, PathMapped, None) + if not access.check("r"): + logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r", PathMapped, user) + return httputils.NOT_ALLOWED + if Conversion is None: Conversion = "none" @@ -902,12 +909,6 @@ class BaseSharing: HiddenByUser = Hidden if ShareType == "token": - # check access Permissions - access = Access(self._rights, user, PathMapped) - if not access.check("r"): - logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r", PathMapped, user) - return httputils.NOT_ALLOWED - if self.permit_create_token is False: if "t" not in access.permissions: logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=False but explict grant misses 't')", PathMapped, user) @@ -971,12 +972,6 @@ class BaseSharing: logger.warning(api_info + ": share already exists with PathMapped=%r User=%r Conversion=%r", PathMapped, User, Conversion) return httputils.CONFLICT - # check access Permissions - access = Access(self._rights, user, PathMapped, None) # PathMapped is mandatory - if not access.check("r") and "i" not in access.permissions: - logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r", PathMapped, user) - return httputils.NOT_ALLOWED - if self.permit_create_map is False: if "m" not in access.permissions: logger.warning(api_info + ": access to PathMapped=%r not allowed for owner %r (permit=False but explicit grant misses 'm')", PathMapped, user) @@ -1041,7 +1036,7 @@ class BaseSharing: else: answer['PathOrToken'] = token - logger.info(api_info + " success: PathMapped=%r Permissions=%r PathOrToken=%r", PathMapped, Permissions, PathOrToken) + logger.notice(api_info + " success: PathMapped=%r Permissions=%r PathOrToken=%r", PathMapped, Permissions, PathOrToken) # action: update elif action == "update": @@ -1173,6 +1168,8 @@ class BaseSharing: logger.warning(api_info + ": %r not successful", request_data['PathOrToken']) return httputils.bad_request("Internal Error") + logger.notice(api_info + " success: PathMapped=%r PathOrToken=%r", PathMapped, PathOrToken) + # action: delete elif action == "delete": logger.trace("" + api_info + ": start") @@ -1213,6 +1210,8 @@ class BaseSharing: logger.warning(api_info + ": %r by user %r not successful", request_data['PathOrToken'], request_data['User']) return httputils.bad_request("Internal Error") + logger.notice(api_info + " success: PathMapped=%r PathOrToken=%r", PathMapped, PathOrToken) + # action: info elif action == "info": logger.info(api_info + ": success") @@ -1301,6 +1300,8 @@ class BaseSharing: logger.warning(api_info + ": %r by user %s not successful", request_data['PathOrToken'], user) return httputils.bad_request("Internal Error") + logger.notice(api_info + " success: PathMapped=%r PathOrToken=%r", PathMapped, PathOrToken) + else: # default logger.warning(api_info + ": unsupported action=%r", action) diff --git a/radicale/sharing/csv.py b/radicale/sharing/csv.py index 473daee0..5eb491cd 100644 --- a/radicale/sharing/csv.py +++ b/radicale/sharing/csv.py @@ -400,7 +400,7 @@ class Sharing(sharing.BaseSharing): logger.debug("sharing database load begin: %r", file) self._sharing_cache = [] with self._storage.acquire_lock("r", None): - with open(file, 'r', newline='') as csvfile: + with open(file, 'r', newline='', encoding=self._encoding) as csvfile: reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';') self._lines = 0 for row in reader: @@ -435,7 +435,14 @@ class Sharing(sharing.BaseSharing): if row[fieldname] is None or row[fieldname] == '': row[fieldname] = {} else: - field = row[fieldname].lstrip('"').rstrip('"').replace("'", '"') + field = row[fieldname].lstrip('"').rstrip('"') + logger.trace("json prep quote replacer match (before): %s", field) + field = field.replace('"', '\\"').replace("\\'", "'") # escape " + field = field.replace("{'", '{"') # replace for JSON start {' -> {" + field = field.replace("'}", '"}') # replace for JSON end '} -> "} + field = field.replace("': '", '": "') # replace for JSON entry/value ': ' -> ": " + field = field.replace("', '", '", "') # replace for JSON delimiter ', ' -> ", " + logger.trace("json prep quote replacer match (after) : %s", field) try: row[fieldname] = json.loads(field) except Exception as e: @@ -453,7 +460,7 @@ class Sharing(sharing.BaseSharing): return True def _write_csv(self, file: str) -> bool: - with open(file, 'w', newline='') as csvfile: + with open(file, 'w', newline='', encoding=self._encoding) as csvfile: writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';') writer.writerows(self._sharing_cache) return True diff --git a/radicale/storage/__init__.py b/radicale/storage/__init__.py index 21b9a24a..b6182a03 100644 --- a/radicale/storage/__init__.py +++ b/radicale/storage/__init__.py @@ -302,6 +302,11 @@ class BaseCollection: class BaseStorage: + _is_collision_free: bool = False + _supports_unicode: bool = False + _supports_trailing_whitespace: bool = False + _supports_problematic_chars: bool = False + def __init__(self, configuration: "config.Configuration") -> None: """Initialize BaseStorage. diff --git a/radicale/storage/multifilesystem/__init__.py b/radicale/storage/multifilesystem/__init__.py index 65ea9b81..54009d69 100644 --- a/radicale/storage/multifilesystem/__init__.py +++ b/radicale/storage/multifilesystem/__init__.py @@ -173,16 +173,19 @@ class Storage( self._makedirs_synced(self._get_collection_root_folder()) logger.info("Storage location subfolder permissions: %s", pathutils.path_permissions_as_string(self._get_collection_root_folder())) logger.info("Storage location subfolder softlink support: %s", pathutils.path_supports_symlink(self._get_collection_root_folder())) - filesystem_root_folder_is_collision_free_case_sensitive = pathutils.path_is_collision_free_case_sensitive(self._get_collection_root_folder()) - filesystem_root_folder_is_collision_free_no_short_filename = pathutils.path_is_collision_free_no_short_filename(self._get_collection_root_folder()) - self._filesystem_root_folder_is_collision_free = filesystem_root_folder_is_collision_free_case_sensitive and filesystem_root_folder_is_collision_free_no_short_filename + is_collision_free_case_sensitive = pathutils.path_is_collision_free_case_sensitive(self._get_collection_root_folder()) + is_collision_free_no_short_filename = pathutils.path_is_collision_free_no_short_filename(self._get_collection_root_folder()) + self._is_collision_free = is_collision_free_case_sensitive and is_collision_free_no_short_filename + self._supports_unicode = pathutils.path_supports_unicode(self._get_collection_root_folder()) + self._supports_trailing_whitespace = pathutils.path_supports_trailing_whitespace(self._get_collection_root_folder()) + self._supports_problematic_chars = pathutils.path_supports_problematic_chars(self._get_collection_root_folder()) logger.info("Storage location subfolder is collision free: %s (case-sensitive=%s no-short-filename=%s)", - self._filesystem_root_folder_is_collision_free, - filesystem_root_folder_is_collision_free_case_sensitive, - filesystem_root_folder_is_collision_free_no_short_filename) - logger.info("Storage location subfolder suppports unicode: %s", pathutils.path_supports_unicode(self._get_collection_root_folder())) - logger.info("Storage location subfolder suppports trailing whitespace: %s", pathutils.path_supports_trailing_whitespace(self._get_collection_root_folder())) - logger.info("Storage location subfolder suppports problematic chars: %s", pathutils.path_supports_problematic_chars(self._get_collection_root_folder())) + self._is_collision_free, + is_collision_free_case_sensitive, + is_collision_free_no_short_filename) + logger.info("Storage location subfolder supports unicode: %s", self._supports_unicode) + logger.info("Storage location subfolder supports trailing whitespace: %s", self._supports_trailing_whitespace) + logger.info("Storage location subfolder supports problematic chars: %s", self._supports_problematic_chars) logger.info("Storage cache subfolder usage for 'item': %s", self._use_cache_subfolder_for_item) logger.info("Storage cache subfolder usage for 'history': %s", self._use_cache_subfolder_for_history) logger.info("Storage cache subfolder usage for 'sync-token': %s", self._use_cache_subfolder_for_synctoken) diff --git a/radicale/storage/multifilesystem/base.py b/radicale/storage/multifilesystem/base.py index 04ce68a7..a17209a0 100644 --- a/radicale/storage/multifilesystem/base.py +++ b/radicale/storage/multifilesystem/base.py @@ -32,7 +32,6 @@ class CollectionBase(storage.BaseCollection): _path: str _encoding: str _filesystem_path: str - _filesystem_root_folder_is_collision_free: bool def __init__(self, storage_: "multifilesystem.Storage", path: str, filesystem_path: Optional[str] = None) -> None: @@ -43,9 +42,12 @@ class CollectionBase(storage.BaseCollection): self._path = pathutils.strip_path(path) self._encoding = storage_.configuration.get("encoding", "stock") self._skip_broken_item = storage_.configuration.get("storage", "skip_broken_item") - self._filesystem_root_folder_is_collision_free = storage_._filesystem_root_folder_is_collision_free + self._is_collision_free = storage_._is_collision_free + self._supports_unicode = storage_._supports_unicode + self._supports_trailing_whitespace = storage_._supports_trailing_whitespace + self._supports_problematic_chars = storage_._supports_problematic_chars if filesystem_path is None: - filesystem_path = pathutils.path_to_filesystem(folder, self.path, self._filesystem_root_folder_is_collision_free) + filesystem_path = pathutils.path_to_filesystem(folder, self.path, self._is_collision_free) self._filesystem_path = filesystem_path # TODO: better fix for "mypy" @@ -81,7 +83,6 @@ class StorageBase(storage.BaseStorage): _folder_umask: str _config_umask: int _max_resource_size: int - _filesystem_root_folder_is_collision_free: bool = False def __init__(self, configuration: config.Configuration) -> None: super().__init__(configuration) diff --git a/radicale/storage/multifilesystem/create_collection.py b/radicale/storage/multifilesystem/create_collection.py index d7e1922c..0e89dad0 100644 --- a/radicale/storage/multifilesystem/create_collection.py +++ b/radicale/storage/multifilesystem/create_collection.py @@ -65,7 +65,7 @@ class StoragePartCreateCollection(StorageBase): # Path should already be sanitized sane_path = pathutils.strip_path(href) - filesystem_path = pathutils.path_to_filesystem(folder, sane_path, self._filesystem_root_folder_is_collision_free) + filesystem_path = pathutils.path_to_filesystem(folder, sane_path, self._is_collision_free) logger.debug("Create collection: %r" % filesystem_path) if not props: diff --git a/radicale/storage/multifilesystem/delete.py b/radicale/storage/multifilesystem/delete.py index cbebdf18..0bdc47fe 100644 --- a/radicale/storage/multifilesystem/delete.py +++ b/radicale/storage/multifilesystem/delete.py @@ -46,7 +46,7 @@ class CollectionPartDelete(CollectionPartHistory, CollectionBase): # Delete an item if not pathutils.is_safe_filesystem_path_component(href): raise pathutils.UnsafePathError(href) - path = pathutils.path_to_filesystem(self._filesystem_path, href, self._filesystem_root_folder_is_collision_free) + path = pathutils.path_to_filesystem(self._filesystem_path, href, self._is_collision_free) if not os.path.isfile(path): raise storage.ComponentNotFoundError(href) os.remove(path) diff --git a/radicale/storage/multifilesystem/discover.py b/radicale/storage/multifilesystem/discover.py index f9542036..ff564b79 100644 --- a/radicale/storage/multifilesystem/discover.py +++ b/radicale/storage/multifilesystem/discover.py @@ -52,7 +52,7 @@ class StoragePartDiscover(StorageBase): # Create the root collection self._makedirs_synced(folder) try: - filesystem_path = pathutils.path_to_filesystem(folder, sane_path, self._filesystem_root_folder_is_collision_free) + filesystem_path = pathutils.path_to_filesystem(folder, sane_path, self._is_collision_free) except ValueError as e: # Path is unsafe logger.warning("Unsafe path %r requested from storage: %s", @@ -110,7 +110,7 @@ class StoragePartDiscover(StorageBase): href = base64.b64encode(group.encode('utf-8')).decode('ascii') logger.debug(f"searching for group calendar {group} {href}") sane_child_path = f"GROUPS/{href}" - if not os.path.isdir(pathutils.path_to_filesystem(folder, sane_child_path, self._filesystem_root_folder_is_collision_free)): + if not os.path.isdir(pathutils.path_to_filesystem(folder, sane_child_path, self._is_collision_free)): continue child_path = f"/GROUPS/{href}/" with child_context_manager(sane_child_path, None): diff --git a/radicale/storage/multifilesystem/get.py b/radicale/storage/multifilesystem/get.py index 670299be..f256885b 100644 --- a/radicale/storage/multifilesystem/get.py +++ b/radicale/storage/multifilesystem/get.py @@ -60,7 +60,7 @@ class CollectionPartGet(CollectionPartCache, CollectionPartLock, raise pathutils.UnsafePathError(href) path = pathutils.path_to_filesystem(self._filesystem_path, href, - self._filesystem_root_folder_is_collision_free) + self._is_collision_free) except ValueError as e: logger.debug( "Can't translate name %r safely to filesystem in %r: %s", diff --git a/radicale/storage/multifilesystem/move.py b/radicale/storage/multifilesystem/move.py index 4c636d43..1febf26c 100644 --- a/radicale/storage/multifilesystem/move.py +++ b/radicale/storage/multifilesystem/move.py @@ -35,8 +35,8 @@ class StoragePartMove(StorageBase): assert isinstance(to_collection, multifilesystem.Collection) assert isinstance(item.collection, multifilesystem.Collection) assert item.href - move_from = pathutils.path_to_filesystem(item.collection._filesystem_path, item.href, self._filesystem_root_folder_is_collision_free) - move_to = pathutils.path_to_filesystem(to_collection._filesystem_path, to_href, self._filesystem_root_folder_is_collision_free) + move_from = pathutils.path_to_filesystem(item.collection._filesystem_path, item.href, self._is_collision_free) + move_to = pathutils.path_to_filesystem(to_collection._filesystem_path, to_href, self._is_collision_free) try: os.replace(move_from, move_to) except OSError as e: diff --git a/radicale/storage/multifilesystem/upload.py b/radicale/storage/multifilesystem/upload.py index f0c25c5e..358b7004 100644 --- a/radicale/storage/multifilesystem/upload.py +++ b/radicale/storage/multifilesystem/upload.py @@ -39,7 +39,7 @@ class CollectionPartUpload(CollectionPartGet, CollectionPartCache, ) -> Tuple[radicale_item.Item, Optional[radicale_item.Item]]: if not pathutils.is_safe_filesystem_path_component(href): raise pathutils.UnsafePathError(href) - path = pathutils.path_to_filesystem(self._filesystem_path, href, self._filesystem_root_folder_is_collision_free) + path = pathutils.path_to_filesystem(self._filesystem_path, href, self._is_collision_free) old_item = self._get(href, verify_href=False) try: with self._atomic_write(path, newline="") as fo: # type: ignore diff --git a/radicale/tests/test_auth.py b/radicale/tests/test_auth.py index 69110e73..d2f71cfe 100644 --- a/radicale/tests/test_auth.py +++ b/radicale/tests/test_auth.py @@ -94,6 +94,14 @@ class TestBaseAuthRequests(BaseTest): def test_htpasswd_plain(self) -> None: self._test_htpasswd("plain", "tmp:bepo") + def test_htpasswd_blacklist_plain(self) -> None: + self._test_htpasswd("plain", "tmp:be:po", ( + ("tm" + chr(9) + "p", "be:po", True), ("tm" + chr(9) + "p", "bepo", False)), check=401) + self._test_htpasswd("plain", "tmp:be:po", ( + ("tm'p", "be:po", True), ("tm'p", "bepo", False)), check=401) + self._test_htpasswd("plain", "tmp:be:po", ( + ('tm"p', "be:po", True), ('tm"p', "bepo", False)), check=401) + def test_htpasswd_plain_autodetect(self) -> None: self._test_htpasswd("autodetect", "tmp:bepo") @@ -103,11 +111,44 @@ class TestBaseAuthRequests(BaseTest): def test_htpasswd_plain_unicode(self) -> None: if not pathutils.path_supports_unicode(self.colpath): - check = 500 + check = 401 else: check = 207 self._test_htpasswd("plain", "😀:🔑", "unicode", check=check) + def test_htpasswd_strict_plain_unicode(self) -> None: + """user with unicode chars is not permitted""" + self.configure({"server": {"validate_user_value": "strict"}}) + if not pathutils.path_supports_unicode(self.colpath): + check = 401 + else: + check = 401 + self._test_htpasswd("plain", "😀:🔑", "unicode", check=check) + + def test_htpasswd_minimal_plain_unicode(self) -> None: + """user with unicode chars is permitted""" + self.configure({"server": {"validate_user_value": "minimal"}}) + if not pathutils.path_supports_unicode(self.colpath): + check = 401 + else: + check = 207 + self._test_htpasswd("plain", "😀:🔑", "unicode", check=check) + + def test_htpasswd_minimal_plain_special(self) -> None: + """user with special chars is not permitted""" + self.configure({"server": {"validate_user_value": "minimal"}}) + check = 401 + self._test_htpasswd("plain", "*?*:bepo", "ascii", check=check) + + def test_htpasswd_unicode_plain_unicode(self) -> None: + """user with unicode symbols is not permitted""" + self.configure({"server": {"validate_user_value": "unicode-letter"}}) + if not pathutils.path_supports_unicode(self.colpath): + check = 401 + else: + check = 401 + self._test_htpasswd("plain", "😀:🔑", "unicode", check=check) + def test_htpasswd_md5(self) -> None: self._test_htpasswd("md5", "tmp:$apr1$BI7VKCZh$GKW4vq2hqDINMr8uv7lDY/") @@ -116,7 +157,7 @@ class TestBaseAuthRequests(BaseTest): def test_htpasswd_md5_unicode(self): if not pathutils.path_supports_unicode(self.colpath): - check = 500 + check = 401 else: check = 207 self._test_htpasswd( @@ -175,7 +216,7 @@ class TestBaseAuthRequests(BaseTest): @pytest.mark.skipif(has_bcrypt == 0, reason="No bcrypt module installed") def test_htpasswd_bcrypt_unicode(self) -> None: if not pathutils.path_supports_unicode(self.colpath): - check = 500 + check = 401 else: check = 207 self._test_htpasswd("bcrypt", "😀:$2y$10$Oyz5aHV4MD9eQJbk6GPemOs4T6edK6U9Sqlzr.W1mMVCS8wJUftnW", "unicode", check=check) @@ -295,16 +336,27 @@ class TestBaseAuthRequests(BaseTest): def test_htpasswd_whitespace_user(self) -> None: for user in (" tmp", "tmp ", " tmp "): if not pathutils.path_supports_trailing_whitespace(self.colpath) and user.endswith(' '): - check = 500 + check = 401 else: check = 207 self._test_htpasswd("plain", "%s:bepo" % user, ( (user, "bepo", True), ("tmp", "bepo", False)), check=check) - def test_htpasswd_problem_user(self) -> None: + def test_htpasswd_problem_user_none(self) -> None: + self.configure({"server": {"validate_user_value": "none"}}) for user in ("tm*p", "tm?p"): if not pathutils.path_supports_problematic_chars(self.colpath): - check = 500 + check = 401 + else: + check = 207 + self._test_htpasswd("plain", "%s:bepo" % user, ( + (user, "bepo", True), ("tmp", "bepo", False)), check=check) + + def test_htpasswd_problem_user_minimal(self) -> None: + self.configure({"server": {"validate_user_value": "none"}}) + for user in ("tm*p", "tm?p"): + if not pathutils.path_supports_problematic_chars(self.colpath): + check = 401 else: check = 207 self._test_htpasswd("plain", "%s:bepo" % user, ( diff --git a/radicale/tests/test_base.py b/radicale/tests/test_base.py index b36497dd..2126f147 100644 --- a/radicale/tests/test_base.py +++ b/radicale/tests/test_base.py @@ -33,7 +33,7 @@ import defusedxml.ElementTree as DefusedET import pytest import vobject -from radicale import storage, utils, xmlutils +from radicale import pathutils, storage, utils, xmlutils from radicale.tests import RESPONSES, BaseTest from radicale.tests.helpers import get_file_content @@ -623,6 +623,56 @@ permissions: RrWw""") self.get(path1, check=404) self.get(path2) + def test_move_unicode(self) -> None: + """Move a item.""" + self.mkcalendar("/calendar.ics/") + event = get_file_content("event1.ics") + path1 = "/calendar.ics/event😀1.ics" + path2 = "/calendar.ics/event😁2.ics" + if not pathutils.path_supports_unicode(self.colpath): + check_put = 400 + check_move = 400 + check_get1 = 400 + check_get2 = 400 + else: + check_put = 201 + check_move = 201 + check_get1 = 200 + check_get2 = 404 + self.put(path1, event, check=check_put) + self.get(path1, check=check_get1) + self.get(path2, check=check_get2) + self.request("MOVE", path1, check=check_move, + HTTP_DESTINATION="http://127.0.0.1/"+path2) + self.get(path1, check=check_get2) + self.get(path2, check=check_get1) + + def test_move_strict_unicode_dst(self) -> None: + """Move a item.""" + self.configure({"server": {"validate_path_value": "strict"}}) + self.mkcalendar("/calendar.ics/") + event = get_file_content("event1.ics") + path1 = "/calendar.ics/event1.ics" + path2 = "/calendar.ics/event😁2.ics" + self.put(path1, event) + self.request("MOVE", path1, check=400, + HTTP_DESTINATION="http://127.0.0.1/"+path2) + self.get(path1, check=200) + self.get(path2, check=400) + + def test_move_strict_unicode_src(self) -> None: + """Move a item.""" + self.configure({"server": {"validate_path_value": "strict"}}) + self.mkcalendar("/calendar.ics/") + event = get_file_content("event1.ics") + path1 = "/calendar.ics/event😀1.ics" + path2 = "/calendar.ics/event2.ics" + self.put(path1, event, check=400) + self.request("MOVE", path1, check=400, + HTTP_DESTINATION="http://127.0.0.1/"+path2) + self.get(path1, check=400) + self.get(path2, check=404) + def test_move_between_collections(self) -> None: """Move a item.""" self.mkcalendar("/calendar1.ics/") diff --git a/radicale/tests/test_sharing.py b/radicale/tests/test_sharing.py index 3302680a..aa355c21 100644 --- a/radicale/tests/test_sharing.py +++ b/radicale/tests/test_sharing.py @@ -26,9 +26,12 @@ import logging import os import re import sys +import tempfile from typing import Dict, Sequence, Tuple, Union -from radicale import sharing, xmlutils +import pytest + +from radicale import pathutils, sharing, xmlutils from radicale.tests import BaseTest from radicale.tests.helpers import get_file_content @@ -45,6 +48,7 @@ class TestSharingApiSanity(BaseTest): encoding: str = self.configuration.get("encoding", "stock") htpasswd = ["owner:ownerpw", "user:userpw", "owner1:owner1pw", "user1:user1pw", + "us😀er:user😀pw", "owner2:owner2pw", "user2:user2pw"] htpasswd_content = "\n".join(htpasswd) with open(self.htpasswd_file_path, "w", encoding=encoding) as f: @@ -1391,6 +1395,7 @@ class TestSharingApiSanity(BaseTest): json_dict: dict path_shared = "/user/calendarUP-shared-by-owner.ics/" + path_shared2 = "/user/calendarUP-shared-by-owner2.ics/" path_mapped = "/owner/calendarUP.ics/" path_mapped2 = "/owner/calendarUP2.ics/" path_mapped_o2 = "/owner2/calendarUP3.ics/" @@ -1425,6 +1430,12 @@ class TestSharingApiSanity(BaseTest): json_dict['PathOrToken'] = path_shared _, headers, answer = self._sharing_api_json("map", "update", check=200, login="owner:ownerpw", json_dict=json_dict) + logging.info("\n*** update map by owner: PathOrToken (json->json) -> 404 (is primary key, therefore not found)") + json_dict = {} + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared2 + _, headers, answer = self._sharing_api_json("map", "update", check=404, login="owner:ownerpw", json_dict=json_dict) + logging.info("\n*** update map by owner: PathMapped(owner2) (json->json) -> 403") json_dict = {} json_dict['PathMapped'] = path_mapped_o2 @@ -3308,6 +3319,8 @@ permissions: RrWw""") json_dict: dict path_owner1 = "/owner1/calendarPGo1.ics/" + path_owner1 = "/owner1/calendarPGo1.ics/" + path_owner2 = "/owner2/calendarPGo1.ics/" path_owner1_rw = "/owner1/calendarPGo1rw.ics/" path_owner1_RrWw = "/owner1/calendarPGo1RrWw.ics/" path_user1_r = "/user1/calendarPGu1-r.ics/" @@ -3326,6 +3339,15 @@ permissions: RrWw""") # create map self.configure({"sharing": {"default_permissions_create_map": "r"}}) + logging.info("\n*** create map user1/owner1 with path of owner 2-> 403") + json_dict = {} + json_dict['User'] = "user1" + json_dict['PathMapped'] = path_owner2 + json_dict['PathOrToken'] = path_user1_r + json_dict['Enabled'] = True + json_dict['Hidden'] = False + _, headers, answer = self._sharing_api_json("map", "create", check=403, login="owner1:owner1pw", json_dict=json_dict) + logging.info("\n*** create map user1/owner1 r -> 200") json_dict = {} json_dict['User'] = "user1" @@ -5400,3 +5422,310 @@ permissions: RrWw""") json_dict['Hidden'] = False json_dict['Conversion'] = "bday" _, headers, answer = self._sharing_api_json("token", "create", check=405, login="owner:ownerpw", json_dict=json_dict) + + def test_sharing_api_map_properies_overlay_unicode(self) -> None: + """share-by-map API usage tests related to properties overlay using unicode.""" + 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"}}) + + json_dict: dict + + logging.info("\n*** prepare and test access") + + 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_mapped = "/owner/calendarPFP-" + db_type + ".ics/" + path_shared_r = "/user/calendarPFP-shared-by-owner-r-" + db_type + ".ics/" + self.mkcalendar(path_mapped, login="owner:ownerpw") + + # 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/" + + description_owner = "Test-Uni😀code-Single'Quote-UmÄlaut-Double\"Quote" + description_user = 'Test-Uni😁code-Single\'Quote-Sßz-Double"quote' + + # execute PROPPATCH as owner + logging.info("\n*** PROPPATCH collection owner -> ok") + self._proppatch_calendar_description(path_mapped, login="owner:ownerpw", description=description_owner) + + # verify PROPPATCH by owner + logging.info("\n*** PROPFIND collection owner (verify collection change) -> ok") + description = self._propfind_calendar_description(path_mapped, login="owner:ownerpw") + assert description == description_owner + + # create map + logging.info("\n*** create map user/owner:rP -> ok") + json_dict = {} + json_dict['User'] = "user" + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared_r + json_dict['Permissions'] = "rP" + 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 PROPFIND as user + logging.info("\n*** PROPFIND collection user") + description = self._propfind_calendar_description(path_shared_r, login="user:userpw") + assert description == description_owner + + # execute PROPPATCH as user + logging.info("\n*** PROPPATCH collection user -> ok") + self._proppatch_calendar_description(path_shared_r, login="user:userpw", description=description_user) + self._proppatch_calendar_color(path_shared_r, login="user:userpw", color="#FFFFFF") + + 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 + + # verify overlay as user + logging.info("\n*** PROPFIND collection user (overlay) -> ok") + description = self._propfind_calendar_description(path_shared_r, login="user:userpw") + assert description == description_user + + # verify overlay not visible by owner + logging.info("\n*** PROPFIND collection owner (no collection change) -> ok") + description = self._propfind_calendar_description(path_mapped, login="owner:ownerpw") + assert description == description_owner + + # check properties file + collection_props_path = os.path.join(self.colpath, "collection-root", path_mapped.removeprefix('/'), ".Radicale.props") + logging.info("collection_props path: %r", collection_props_path) + with open(collection_props_path) as f: + props = json.load(f) + logging.info("collection_props: %r", props) + assert props['C:calendar-description'] == description_owner + + # reconfigure to trigger restart and reparsing of database + self.configure({"auth": {"type": "htpasswd"}}) + + # verify overlay as user + logging.info("\n*** PROPFIND collection user (overlay) -> ok") + description = self._propfind_calendar_description(path_shared_r, login="user:userpw") + assert description == description_user + + @pytest.mark.skipif(not pathutils.path_supports_unicode(tempfile.mkdtemp()), reason="TEMP is not supporting unicode") + def test_sharing_api_map_user_unicode(self) -> None: + """share-by-map API usage tests related to properties overlay using unicode.""" + 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"}}) + + json_dict: dict + + logging.info("\n*** prepare and test access") + + 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_mapped = "/owner/calendarPFP-" + db_type + ".ics/" + path_shared_r = "/us😀er/calendarPFP-shared-by-owner-r-" + db_type + ".ics/" + self.mkcalendar(path_mapped, login="owner:ownerpw") + + # create map + logging.info("\n*** create map user/owner:rP -> ok") + json_dict = {} + json_dict['User'] = "us😀er" + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared_r + json_dict['Permissions'] = "rP" + 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'] = "us😀er" + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared_r + _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="us😀er:user😀pw", json_dict=json_dict) + + @pytest.mark.skipif(not pathutils.path_supports_unicode(tempfile.mkdtemp()), reason="TEMP is not supporting unicode") + def test_sharing_api_map_path_unicode(self) -> None: + """share-by-map API usage tests related to properties overlay using unicode.""" + 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"}}) + + json_dict: dict + + logging.info("\n*** prepare and test access") + + 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_mapped = "/owner/calendar😀PFP-" + db_type + ".ics/" + path_shared_r = "/user/calendar😁PFP-shared-by-owner-r-" + db_type + ".ics/" + self.mkcalendar(path_mapped, login="owner:ownerpw") + + # create map + logging.info("\n*** create map user/owner:rP -> ok") + json_dict = {} + json_dict['User'] = "user" + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared_r + json_dict['Permissions'] = "rP" + 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) + + def test_sharing_api_map_strict_user_unicode(self) -> None: + """share-by-map API usage tests related to properties overlay using unicode in user.""" + 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"}, + "server": {"validate_user_value": "strict"}, + "rights": {"type": "owner_only"}}) + + json_dict: dict + + logging.info("\n*** prepare and test access") + + 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_mapped = "/owner/calendarPFP-" + db_type + ".ics/" + path_shared_r = "/user/calendarPFP-shared-by-owner-r-" + db_type + ".ics/" + self.mkcalendar(path_mapped, login="owner:ownerpw") + + # create map + logging.info("\n*** create map user/owner:rP -> failed") + json_dict = {} + json_dict['User'] = "us😁er" + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared_r + json_dict['Permissions'] = "rP" + json_dict['Enabled'] = True + json_dict['Hidden'] = False + _, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict) + + def test_sharing_api_map_strict_path_unicode(self) -> None: + """share-by-map API usage tests related to properties overlay using unicode in user.""" + 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"}, + "server": {"validate_path_value": "strict"}, + "rights": {"type": "owner_only"}}) + + json_dict: dict + + logging.info("\n*** prepare and test access") + + for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)): + logging.info("\n*** test: %s", db_type) + self.configure({"sharing": {"type": db_type}}) + + logging.info("\n*** create collection, already rejected in early state") + path_mapped = "/owner/calendar😀PFP-" + db_type + ".ics/" + path_shared_r = "/user/calendarPFP-shared-by-owner-r-" + db_type + ".ics/" + self.mkcalendar(path_mapped, login="owner:ownerpw", check=400) + + logging.info("\n*** create collection") + path_mapped = "/owner/calendarPFP-" + db_type + ".ics/" + path_shared_r = "/user/calendar😁PFP-shared-by-owner-r-" + db_type + ".ics/" + self.mkcalendar(path_mapped, login="owner:ownerpw") + + # create map + logging.info("\n*** create map user/owner:rP -> ok") + json_dict = {} + json_dict['User'] = "user" + json_dict['PathMapped'] = path_mapped + json_dict['PathOrToken'] = path_shared_r + json_dict['Permissions'] = "rP" + json_dict['Enabled'] = True + json_dict['Hidden'] = False + _, headers, answer = self._sharing_api_json("map", "create", check=400, login="owner:ownerpw", json_dict=json_dict)