Avii/Radicale
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

sharing/via-proxy: add support and test cases

Browse Source
This commit is contained in:
Peter Bieringer
2026-03-24 12:26:18 +01:00
parent 0fae7c5a7c
commit d0106c1a6b
3 changed files with 91 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
radicale/sharing/__init__.py
View File

@@ -779,7 +779,11 @@ class BaseSharing:
# check for optional parameters
if 'PathMapped' in request_data:
# used by create or list(filter)
PathMapped = request_data['PathMapped']
if base_prefix:
PathMapped = request_data['PathMapped'].removeprefix(base_prefix)
logger.debug(api_info + ": remove base_prefix PathMapped=%r->%r", request_data['PathMapped'], PathMapped)
else:
PathMapped = request_data['PathMapped']
if 'PathOrToken' not in request_data:
if action == 'info':
@@ -1080,7 +1084,10 @@ class BaseSharing:
if ShareType == "token":
PathOrToken = token
answer['PathOrToken'] = token
if base_prefix:
answer['PathOrToken'] = base_prefix + token
else:
answer['PathOrToken'] = token
logger.info(api_info + " success: PathMapped=%r Permissions=%r PathOrToken=%r", PathMapped, Permissions, PathOrToken)

3
radicale/tests/__init__.py
View File

@@ -90,6 +90,7 @@ class BaseTest:
remote_useragent = kwargs.pop("remote_useragent", None)
remote_host = kwargs.pop("remote_host", None)
content_type = kwargs.pop("content_type", None)
x_forwarded_for = kwargs.pop("x_forwarded_for", None)
accept = kwargs.pop("accept", None)
environ: Dict[str, Any] = {k.upper(): v for k, v in kwargs.items()}
for k, v in environ.items():
@@ -108,6 +109,8 @@ class BaseTest:
environ["REMOTE_ADDR"] = remote_host
if content_type:
environ["CONTENT_TYPE"] = content_type
if x_forwarded_for:
environ["HTTP_X_FORWARDED_FOR"] = x_forwarded_for
if accept:
environ["HTTP_ACCEPT"] = accept
environ["REQUEST_METHOD"] = method.upper()

87
radicale/tests/test_sharing.py
View File

@@ -49,31 +49,39 @@ class TestSharingApiSanity(BaseTest):
f.write(htpasswd_content)
# Helper functions
def _sharing_api(self, sharing_type: str, action: str, check: int, login: Union[str, None], data: str, content_type: str, accept: Union[str, None]) -> Tuple[int, Dict[str, str], str]:
def _sharing_api(self, sharing_type: str, action: str, check: int, login: Union[str, None], data: str, content_type: str, accept: Union[str, None], prefix: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
path_base = "/.sharing/v1/" + sharing_type + "/"
_, headers, answer = self.request("POST", path_base + action, check=check, login=login, data=data, content_type=content_type, accept=accept)
if prefix is not None:
path_base = prefix + path_base
_, headers, answer = self.request("POST", path_base + action, check=check, login=login, data=data, content_type=content_type, accept=accept, x_forwarded_for="127.0.0.2")
else:
_, headers, answer = self.request("POST", path_base + action, check=check, login=login, data=data, content_type=content_type, accept=accept)
logging.info("received answer:\n%s", "\n".join(answer.splitlines()))
return _, headers, answer
def _sharing_api_form(self, sharing_type: str, action: str, check: int, login: Union[str, None], form_array: Sequence[str], accept: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
def _sharing_api_form(self, sharing_type: str, action: str, check: int, login: Union[str, None], form_array: Sequence[str], accept: Union[str, None] = None, prefix: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
data = "&".join(form_array)
content_type = "application/x-www-form-urlencoded"
if accept is None:
accept = "text/plain"
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept)
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept, prefix=prefix)
return _, headers, answer
def _sharing_api_json(self, sharing_type: str, action: str, check: int, login: Union[str, None], json_dict: dict, accept: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
def _sharing_api_json(self, sharing_type: str, action: str, check: int, login: Union[str, None], json_dict: dict, accept: Union[str, None] = None, prefix: Union[str, None] = None) -> Tuple[int, Dict[str, str], str]:
data = json.dumps(json_dict)
content_type = "application/json"
if accept is None:
accept = "application/json"
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept)
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept, prefix=prefix)
return _, headers, answer
def _propfind_allprop(self, path: str, login: str = "") -> dict:
def _propfind_allprop(self, path: str, login: str = "", prefix: Union[str, None] = None) -> dict:
propfind_allprop = get_file_content("allprop.xml")
_, responses = self.propfind(path=path, data=propfind_allprop, login=login)
if prefix is not None:
path = prefix + path
_, responses = self.propfind(path=path, data=propfind_allprop, login=login, x_forwarded_for="127.0.0.2")
else:
_, responses = self.propfind(path=path, data=propfind_allprop, login=login)
logging.info("response: %r", responses)
response = responses[path]
assert not isinstance(response, int)
@@ -679,6 +687,69 @@ class TestSharingApiSanity(BaseTest):
self.delete(path_base1, login="owner:ownerpw")
self.delete(path_base2, login="owner:ownerpw")
def test_sharing_api_token_usage_proxy(self) -> None:
"""share-by-token API tests simulating a reverse proxy - real usage."""
script_name = "/radicale"
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": "True",
"response_content_on_debug": "True",
"request_content_on_debug": "True"},
"server": {"script_name": script_name},
"rights": {"type": "owner_only"}})
json_dict: dict
path_base = "/owner/calendar.ics/"
event = get_file_content("event1.ics")
path = path_base + "/event1.ics"
logging.info("\n*** prepare")
self.mkcalendar(path_base, login="owner:ownerpw")
self.put(path, event, login="owner:ownerpw")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
logging.info("\n*** test access to collection")
_, headers, answer = self.request("GET", path_base, check=200, login="owner:ownerpw")
assert "UID:event" in answer
logging.info("\n*** test access to item")
_, headers, answer = self.request("GET", path, check=200, login="owner:ownerpw")
assert "UID:event" in answer
logging.info("\n*** create token")
json_dict = {}
json_dict["PathMapped"] = script_name + path_base
json_dict["Enabled"] = True
json_dict["Hidden"] = False
_, headers, answer = self._sharing_api_json("token", "create", check=200, login="owner:ownerpw", json_dict=json_dict, prefix=script_name)
answer_dict = json.loads(answer)
assert "Status" in answer_dict
assert "PathOrToken" in answer_dict
Token = answer_dict["PathOrToken"]
logging.debug("Token: %r", Token)
assert Token.startswith(script_name) is True
path_shared = Token
# check PROPFIND item as owner (remove prefix again as added later)
logging.info("\n*** PROPFIND item as owner -> calendar")
response = self._propfind_allprop(path_shared.removeprefix(script_name), login="owner:ownerpw", prefix=script_name)
logging.debug("response: %r", response)
assert "CR:supported-address-data" not in response
assert "C:supported-calendar-component-set" in response
assert "D:current-user-privilege-set" in response
def test_sharing_api_token_usage(self) -> None:
"""share-by-token API tests - real usage."""
self.configure({"auth": {"type": "htpasswd",