Merge pull request #2098 from pbiering/regression-fixes

Regression fixes
This commit is contained in:
Peter Bieringer
2026-04-23 12:26:35 +02:00
committed by GitHub
26 changed files with 745 additions and 84 deletions

View File

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

View File

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

View File

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

9
config
View File

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

View File

@@ -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" ]

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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",

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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",

View File

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

View File

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

View File

@@ -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, (

View File

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

View File

@@ -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, """\
<?xml version="1.0" encoding="utf-8"?>
<propfind xmlns="DAV:">
<prop>
<current-user-principal />
</prop>
</propfind>""", 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)