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

Merge pull request #2124 from pbiering/logging-conditional

Browse Source 
Add support for conditional logging
This commit is contained in:
Peter Bieringer
2026-05-12 09:17:07 +02:00
committed by GitHub
parent 66de9aab36 aefa99d5a4
commit d7f763f05f
25 changed files with 658 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
DOCUMENTATION.md
View File

@@ -1803,6 +1803,65 @@ Log response content (body) on `level = debug`
Default: `False`
##### request_header_on_notice_condition
_(>= 3.7.3)_
Log request header on `level = notice` if condition is fulfilled
Default: `{}`
Format: JSON structure as text
Supported tokens:
* `method`
* `path`
* `useragent`
* `host` (IPv4/IPv6 address/network)
* `login`
* `status` (only supported on responses)
Supported matches:
* str: `startswith`, `endswith`, `equal`, `re`
* int: `==`, `<`, `<=`, `>=`, `>`, `!=`
* IP address: `equal`, `==`, `!=`
* IP network: `included`, `excluded`
Examples:
* `{"method": {"match": "equal", "value": "GET"}, "login": {"match": "equal", "value": "owner"}}`
* `{"method": {"match": "equal", "value": "PUT"}, "status": {"match": ">=", "value": "201"}, "host": {"match": "included", "value": "127.0.0.0/8"}}`
* `{"method": {"match": "equal", "value": "PUT"}, "path": {"match": "re", "value": "ev?nt[01]"}}`
##### request_content_on_notice_condition
_(>= 3.7.3)_
Log request content (body) on `level = notice` if condition is fulfilled
Default: `{}`
Format: see `request_header_on_notice_condition`
##### response_header_on_notice_condition
_(>= 3.7.3)_
Log response header on `level = notice` if condition is fulfilled
Default: `{}`
Format: see `request_header_on_notice_condition`
##### response_content_on_notice_condition
_(>= 3.7.3)_
Log response content (body) on `level = notice` if condition is fulfilled
Default: `{}`
Format: see `request_header_on_notice_condition`
##### rights_rule_doesnt_match_on_debug
_(>= 3.2.3)_

12
config
View File

@@ -397,6 +397,18 @@
# Log response content on level=debug
#response_content_on_debug = False
# Log request header on level=notice with condition
#request_header_on_notice_condition = {}
# Log request content on level=notice with condition
#request_content_on_notice_condition = {}
# Log response header on level=notice with condition
#response_header_on_notice_condition = {}
# Log response content on level=notice with condition
#response_content_on_notice_condition = {}
# Log rights rule which doesn't match on level=debug
#rights_rule_doesnt_match_on_debug = False

67
radicale/app/__init__.py
View File

@@ -130,6 +130,14 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
logger.debug("log request content on debug: %s", self._request_content_on_debug)
logger.debug("log response header on debug: %s", self._response_header_on_debug)
logger.debug("log response content on debug: %s", self._response_content_on_debug)
self._request_header_on_notice_condition = configuration.get("logging", "request_header_on_notice_condition")
self._request_content_on_notice_condition = configuration.get("logging", "request_content_on_notice_condition")
self._response_header_on_notice_condition = configuration.get("logging", "response_header_on_notice_condition")
self._response_content_on_notice_condition = configuration.get("logging", "response_content_on_notice_condition")
logger.notice("log request header on notice condition: %s", self._request_header_on_notice_condition)
logger.notice("log request content on notice condition: %s", self._request_content_on_notice_condition)
logger.notice("log response header on notice condition: %s", self._response_header_on_notice_condition)
logger.notice("log response content on notice condition: %s", self._response_content_on_notice_condition)
self._limit_content = configuration.get("logging", "limit_content")
logger.debug("log limit for content: %d", self._limit_content)
self._auth_delay = configuration.get("auth", "delay")
@@ -274,7 +282,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
"""Manage a request."""
def response(status: int, headers: types.WSGIResponseHeaders,
answer: Union[None, str, bytes],
xml_request: Union[None, str] = None) -> _IntermediateResponse:
xml_request: Union[None, str] = None, request_info: dict = {}) -> _IntermediateResponse:
"""Helper to create response from internal types.WSGIResponse"""
headers = dict(headers)
content_encoding = "plain"
@@ -286,8 +294,16 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content (nonXML):\n%s", utils.textwrap_str(answer, self._limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content: suppressed by config/option [logging] response_content_on_debug")
if self._response_content_on_notice_condition != {}:
if log.log_conditional(
"response-content",
condition=self._response_content_on_notice_condition,
value=request_info,
):
logger.notice("Response content (nonXML, log condition passed):\n%s", utils.textwrap_str(answer, self._limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content: suppressed by config/option [logging] response_content_on_debug")
headers["Content-Type"] += "; charset=%s" % self._encoding
answer = answer.encode(self._encoding)
accept_encoding = [
@@ -307,12 +323,22 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
# Add extra headers set in configuration
headers.update(self._extra_headers)
request_info["status"] = status
if self._response_header_on_debug:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response header:\n%s", utils.textwrap_str(pprint.pformat(headers), self._limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response header: suppressed by config/option [logging] response_header_on_debug")
if self._response_header_on_notice_condition != {}:
if log.log_conditional(
"response-header",
condition=self._response_header_on_notice_condition,
value=request_info,
):
logger.notice("Response header (log condition passed):\n%s", utils.textwrap_str(pprint.pformat(headers), self._limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response header: suppressed by config/option [logging] response_header_on_debug")
# Start response
# delay on error
@@ -428,8 +454,10 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
if environ.get("HTTP_X_FORWARDED_HOST") or environ.get("HTTP_X_FORWARDED_PROTO") or environ.get("HTTP_X_FORWARDED_SERVER"):
reverse_proxy = True
remote_useragent = ""
remote_useragent_txt = ""
if environ.get("HTTP_USER_AGENT"):
remote_useragent = " using %r" % environ["HTTP_USER_AGENT"]
remote_useragent = environ["HTTP_USER_AGENT"]
remote_useragent_txt = " using %r" % environ["HTTP_USER_AGENT"]
depthinfo = ""
if environ.get("HTTP_DEPTH"):
depthinfo = " with depth %r" % environ["HTTP_DEPTH"]
@@ -439,12 +467,14 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
https_info = ""
logger.info("%s request for %r%s received from %s%s%s",
request_method, unsafe_path, depthinfo,
remote_host, remote_useragent, https_info)
remote_host, remote_useragent_txt, https_info)
if self._request_header_on_debug:
logger.debug("Request header:\n%s",
utils.textwrap_str(pprint.pformat(self._scrub_headers(environ)), self._limit_content))
else:
logger.debug("Request header: suppressed by config/option [logging] request_header_on_debug")
if not self._request_header_on_notice_condition != {}:
# conditional request header logging is later
logger.debug("Request header: suppressed by config/option [logging] request_header_on_debug")
# SCRIPT_NAME is already removed from PATH_INFO, according to the
# WSGI specification.
@@ -527,6 +557,23 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
self._rights._user_groups = self._auth._ldap_groups
except AttributeError:
pass
request_info: dict = {
"method": request_method,
"login": login, # not 'user' in this step
"path": path,
"useragent": remote_useragent,
"host": remote_host,
}
if not self._request_header_on_debug and self._request_header_on_notice_condition != {}:
if log.log_conditional(
"request-header",
condition=self._request_header_on_notice_condition,
value=request_info,
):
logger.notice("Request header (log condition passed):\n%s", utils.textwrap_str(pprint.pformat(self._scrub_headers(environ)), self._limit_content))
if user and login == user:
logger.info("Successful login: %r (%s)", user, info)
elif user:
@@ -611,7 +658,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
try:
status, headers, answer, xml_request = function(
environ, base_prefix, path, user, remote_host, remote_useragent)
environ, base_prefix, path, user, request_info)
except PermissionError as e:
logger.error("PermissionError: %s", e)
status, headers, answer, xml_request = httputils.INTERNAL_SERVER_ERROR
@@ -648,4 +695,4 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
"WWW-Authenticate":
"Basic realm=\"%s\"" % self._auth_realm})
return response(status, headers, answer, xml_request)
return response(status, headers, answer, xml_request, request_info)

50
radicale/app/base.py
View File

@@ -23,7 +23,7 @@ import unicodedata
import xml.etree.ElementTree as ET
from typing import Optional, Union
from radicale import (auth, config, hook, httputils, pathutils, rights,
from radicale import (auth, config, hook, httputils, log, pathutils, rights,
sharing, storage, types, utils, web, xmlutils)
from radicale.log import logger
from radicale.rights import intersect
@@ -137,12 +137,16 @@ class ApplicationBase:
self._log_bad_put_request_content = configuration.get("logging", "bad_put_request_content")
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._response_content_on_notice_condition = configuration.get("logging", "response_content_on_notice_condition")
self._request_content_on_notice_condition = configuration.get("logging", "request_content_on_notice_condition")
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
def _read_xml_request_body(self,
environ: types.WSGIEnviron,
request_info: dict,
) -> Optional[ET.Element]:
content = httputils.decode_request(
self.configuration, environ,
@@ -154,19 +158,41 @@ class ApplicationBase:
except ET.ParseError as e:
logger.debug("Request content (Invalid XML):\n%s", content)
raise RuntimeError("Failed to parse XML: %s" % e) from e
if logger.isEnabledFor(logging.DEBUG):
if self._request_content_on_debug:
logger.debug("Request content (XML):\n%s",
utils.textwrap_str(xmlutils.pretty_xml(xml_content)))
if self._request_content_on_debug:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content (XML):\n%s", utils.textwrap_str(xmlutils.pretty_xml(xml_content)))
else:
if self._request_content_on_notice_condition != {}:
if log.log_conditional(
"request-content",
condition=self._request_content_on_notice_condition,
value=request_info,
):
logger.notice("Request content (XML, log condition passed):\n%s", utils.textwrap_str(xmlutils.pretty_xml(xml_content)))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content (XML, log condition skipped): suppressed")
else:
logger.debug("Request content (XML): suppressed by config/option [logging] request_content_on_debug")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content (XML): suppressed by config/option [logging] request_content_on_debug")
return xml_content
def _xml_response(self, xml_content: ET.Element) -> bytes:
if logger.isEnabledFor(logging.DEBUG):
if self._response_content_on_debug:
def _xml_response(self, xml_content: ET.Element, request_info: dict) -> bytes:
if self._response_content_on_debug:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content (XML):\n%s",
utils.textwrap_str(xmlutils.pretty_xml(xml_content), self._limit_content))
else:
if self._response_content_on_notice_condition != {}:
if log.log_conditional(
"response-content",
condition=self._response_content_on_notice_condition,
value=request_info,
):
logger.notice("Response content (XML, log condition passed):\n%s", utils.textwrap_str(xmlutils.pretty_xml(xml_content)))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content (XML, log condition skipped): suppressed")
else:
logger.debug("Response content (XML): suppressed by config/option [logging] response_content_on_debug")
f = io.BytesIO()
@@ -174,11 +200,11 @@ class ApplicationBase:
xml_declaration=True)
return f.getvalue()
def _webdav_error_response(self, status: int, human_tag: str
def _webdav_error_response(self, status: int, human_tag: str, request_info: dict
) -> types.WSGIResponse:
"""Generate XML error response."""
headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
content = self._xml_response(xmlutils.webdav_error(human_tag))
content = self._xml_response(xmlutils.webdav_error(human_tag), request_info)
return status, headers, content, None

4
radicale/app/delete.py
View File

@@ -55,7 +55,7 @@ def xml_delete(base_prefix: str, path: str, collection: storage.BaseCollection,
class ApplicationPartDelete(ApplicationBase):
def do_DELETE(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage DELETE request."""
actor = user
permissions_filter = None
@@ -126,4 +126,4 @@ class ApplicationPartDelete(ApplicationBase):
for notification_item in hook_notification_item_list: # Will be empty if hook not enabled
self._hook.notify(notification_item)
headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
return client.OK, headers, self._xml_response(xml_answer), None
return client.OK, headers, self._xml_response(xml_answer, request_info), None

4
radicale/app/get.py
View File

@@ -68,7 +68,7 @@ class ApplicationPartGet(ApplicationBase):
return value
def do_GET(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
"""Manage GET request."""
# Redirect to /.web if the root path is requested
if not pathutils.strip_path(path):
@@ -84,7 +84,7 @@ class ApplicationPartGet(ApplicationBase):
base_prefix + unsafe_path, location)
return httputils.redirect(location, client.MOVED_PERMANENTLY)
# Dispatch /.web path to web module
return self._web.get(environ, base_prefix, path, user)
return self._web.get(environ, base_prefix, path, user, request_info)
permissions_filter = None
share = None
if self._sharing._enabled:

4
radicale/app/head.py
View File

@@ -26,7 +26,7 @@ from radicale.app.get import ApplicationPartGet
class ApplicationPartHead(ApplicationPartGet, ApplicationBase):
def do_HEAD(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
"""Manage HEAD request."""
# Body is dropped in `Application.__call__` for HEAD requests
return self.do_GET(environ, base_prefix, path, user, remote_host, remote_useragent)
return self.do_GET(environ, base_prefix, path, user, request_info)

6
radicale/app/mkcalendar.py
View File

@@ -32,12 +32,12 @@ from radicale.log import logger
class ApplicationPartMkcalendar(ApplicationBase):
def do_MKCALENDAR(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage MKCALENDAR request."""
if "w" not in self._rights.authorization(user, path):
return httputils.NOT_ALLOWED
try:
xml_content = self._read_xml_request_body(environ)
xml_content = self._read_xml_request_body(environ, request_info)
except RuntimeError as e:
logger.warning(
"Bad MKCALENDAR request on %r: %s", path, e, exc_info=True)
@@ -67,7 +67,7 @@ class ApplicationPartMkcalendar(ApplicationBase):
item = next(iter(self._storage.discover(path)), None)
if item:
return self._webdav_error_response(
client.CONFLICT, "D:resource-must-be-null")
client.CONFLICT, "D:resource-must-be-null", request_info)
parent_path = pathutils.parent_path(path)
parent_item = next(iter(self._storage.discover(parent_path)), None)
if not parent_item:

4
radicale/app/mkcol.py
View File

@@ -32,13 +32,13 @@ from radicale.log import logger
class ApplicationPartMkcol(ApplicationBase):
def do_MKCOL(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage MKCOL request."""
permissions = self._rights.authorization(user, path)
if not rights.intersect(permissions, "Ww"):
return httputils.NOT_ALLOWED
try:
xml_content = self._read_xml_request_body(environ)
xml_content = self._read_xml_request_body(environ, request_info)
except RuntimeError as e:
logger.warning(
"Bad MKCOL request on %r: %s", path, e, exc_info=True)

4
radicale/app/move.py
View File

@@ -49,7 +49,7 @@ def get_server_netloc(environ: types.WSGIEnviron, force_port: bool = False):
class ApplicationPartMove(ApplicationBase):
def do_MOVE(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage MOVE request."""
raw_dest = environ.get("HTTP_DESTINATION", "")
@@ -136,7 +136,7 @@ class ApplicationPartMove(ApplicationBase):
to_collection.has_uid(item.uid)):
return self._webdav_error_response(
client.CONFLICT, "%s:no-uid-conflict" % (
"C" if collection_tag == "VCALENDAR" else "CR"))
"C" if collection_tag == "VCALENDAR" else "CR"), request_info)
to_href = posixpath.basename(pathutils.strip_path(to_path))
try:
self._storage.move(item, to_collection, to_href)

2
radicale/app/options.py
View File

@@ -28,7 +28,7 @@ from radicale.app.base import ApplicationBase
class ApplicationPartOptions(ApplicationBase):
def do_OPTIONS(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage OPTIONS request."""
headers = {
"Allow": ", ".join(

6
radicale/app/post.py
View File

@@ -26,10 +26,10 @@ from radicale.app.base import ApplicationBase
class ApplicationPartPost(ApplicationBase):
def do_POST(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage POST request."""
if path == "/.web" or path.startswith("/.web/"):
return self._web.post(environ, base_prefix, path, user)
return self._web.post(environ, base_prefix, path, user, request_info)
elif path == "/.sharing" or path.startswith("/.sharing/"):
return self._sharing.post(environ, base_prefix, path, user)
return self._sharing.post(environ, base_prefix, path, user, request_info)
return httputils.METHOD_NOT_ALLOWED

7
radicale/app/propfind.py
View File

@@ -574,7 +574,7 @@ class ApplicationPartPropfind(ApplicationBase):
yield item, permission, raw_permissions
def do_PROPFIND(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage PROPFIND request."""
http_depth = environ.get("HTTP_DEPTH", "0")
permissions_filter = None
@@ -594,7 +594,7 @@ class ApplicationPartPropfind(ApplicationBase):
if not access.check("r"):
return httputils.NOT_ALLOWED
try:
xml_content = self._read_xml_request_body(environ)
xml_content = self._read_xml_request_body(environ, request_info)
except RuntimeError as e:
logger.warning(
"Bad PROPFIND request on %r: %s", path, e, exc_info=True)
@@ -659,4 +659,5 @@ class ApplicationPartPropfind(ApplicationBase):
allowed_items, user, self._encoding, max_resource_size=self._max_resource_size, shares=shares)
if xml_answer is None:
return httputils.NOT_ALLOWED
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)
request_info["status"] = client.MULTI_STATUS
return client.MULTI_STATUS, headers, self._xml_response(xml_answer, request_info), xmlutils.pretty_xml(xml_content)

11
radicale/app/proppatch.py
View File

@@ -97,7 +97,7 @@ def xml_proppatch(base_prefix: str, path: str,
class ApplicationPartProppatch(ApplicationBase):
def do_PROPPATCH(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage PROPPATCH request."""
actor = user
permissions_filter = None
@@ -155,7 +155,8 @@ class ApplicationPartProppatch(ApplicationBase):
logger.info("PROPPATCH request on shared %r: write-permissions, overlay not enforced, but enforced by permission 'E'", path_orig)
share_overlay = True
try:
xml_content = self._read_xml_request_body(environ)
xml_content = self._read_xml_request_body(environ, request_info)
except RuntimeError as e:
logger.warning(
"Bad PROPPATCH request on %r: %s", path, e, exc_info=True)
@@ -193,7 +194,8 @@ class ApplicationPartProppatch(ApplicationBase):
logger.warning(
"Bad PROPPATCH request on %r: %s", path, e, exc_info=True)
return httputils.BAD_REQUEST
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)
request_info["status"] = client.MULTI_STATUS
return client.MULTI_STATUS, headers, self._xml_response(xml_answer, request_info), xmlutils.pretty_xml(xml_content)
with self._storage.acquire_lock("w", user, path=path, request="PROPPATCH"):
item = next(iter(self._storage.discover(path)), None)
@@ -242,4 +244,5 @@ class ApplicationPartProppatch(ApplicationBase):
logger.warning(
"Bad PROPPATCH request on %r: %s", path, e, exc_info=True)
return httputils.BAD_REQUEST
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)
request_info["status"] = client.MULTI_STATUS
return client.MULTI_STATUS, headers, self._xml_response(xml_answer, request_info), xmlutils.pretty_xml(xml_content)

7
radicale/app/put.py
View File

@@ -179,7 +179,7 @@ def prepare(vobject_items: List[vobject.base.Component], path: str,
class ApplicationPartPut(ApplicationBase):
def do_PUT(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage PUT request."""
actor = user
permissions_filter = None
@@ -196,7 +196,7 @@ class ApplicationPartPut(ApplicationBase):
if not access.check("w"):
return httputils.NOT_ALLOWED
try:
content = httputils.read_request_body(self.configuration, environ)
content = httputils.read_request_body(self.configuration, environ, request_info)
except RuntimeError as e:
logger.warning("Bad PUT request on %r (read_request_body): %s", path, e, exc_info=True)
return httputils.BAD_REQUEST
@@ -344,9 +344,10 @@ class ApplicationPartPut(ApplicationBase):
prepared_item, = prepared_items
if (item and item.uid != prepared_item.uid or
not item and parent_item.has_uid(prepared_item.uid)):
request_info["status"] = client.CONFLICT
return self._webdav_error_response(
client.CONFLICT, "%s:no-uid-conflict" % (
"C" if tag == "VCALENDAR" else "CR"))
"C" if tag == "VCALENDAR" else "CR"), request_info)
href = posixpath.basename(pathutils.strip_path(path))
try:

16
radicale/app/report.py
View File

@@ -149,7 +149,7 @@ def free_busy_report(base_prefix: str, path: str, xml_request: Optional[ET.Eleme
def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
collection: storage.BaseCollection, encoding: str,
unlock_storage_fn: Callable[[], None],
max_occurrence: int = 0, user: str = "", remote_addr: str = "", remote_useragent: str = "",
max_occurrence: int = 0, user: str = "", request_info: dict = {},
share: Union[dict, None] = None) -> Tuple[int, ET.Element]:
"""Read and answer REPORT requests that return XML.
@@ -216,8 +216,11 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
sync_token, names = collection.sync(old_sync_token)
except ValueError as e:
# Invalid sync token
remote_useragent_txt = ""
if request_info["useragent"] != "":
remote_useragent_txt = " using %r" % request_info["useragent"]
logger.warning("Client provided invalid sync token for path %r (user %r from %s%s): %s",
path, user, remote_addr, remote_useragent, e, exc_info=True)
path, user, request_info["host"], remote_useragent_txt, e, exc_info=True)
# client.CONFLICT doesn't work with some clients (e.g. InfCloud)
return (client.FORBIDDEN,
xmlutils.webdav_error("D:valid-sync-token"))
@@ -851,7 +854,7 @@ def test_filter(collection_tag: str, item: radicale_item.Item,
class ApplicationPartReport(ApplicationBase):
def do_REPORT(self, environ: types.WSGIEnviron, base_prefix: str,
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
path: str, user: str, request_info: dict) -> types.WSGIResponse:
"""Manage REPORT request."""
permissions_filter = None
share = None
@@ -867,7 +870,7 @@ class ApplicationPartReport(ApplicationBase):
if not access.check("r"):
return httputils.NOT_ALLOWED
try:
xml_content = self._read_xml_request_body(environ)
xml_content = self._read_xml_request_body(environ, request_info)
except RuntimeError as e:
logger.warning("Bad REPORT request on %r: %s", path, e,
exc_info=True)
@@ -905,10 +908,11 @@ class ApplicationPartReport(ApplicationBase):
try:
status, xml_answer = xml_report(
base_prefix, path, xml_content, collection, self._encoding,
lock_stack.close, max_occurrence, user, remote_host, remote_useragent, share=share)
lock_stack.close, max_occurrence, user, request_info, share=share)
except ValueError as e:
logger.warning(
"Bad REPORT request on %r: %s", path, e, exc_info=True)
return httputils.BAD_REQUEST
headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
return status, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)
request_info["status"] = status
return status, headers, self._xml_response(xml_answer, request_info), xmlutils.pretty_xml(xml_content)

81
radicale/config.py
View File

@@ -27,9 +27,11 @@ Use ``load()`` to obtain an instance of ``Configuration`` for use with
"""
import contextlib
import ipaddress
import json
import math
import os
import re
import string
import sys
from collections import OrderedDict
@@ -167,6 +169,66 @@ def json_str(value: Any) -> dict:
return ret
def json_str_condition(value: Any) -> dict:
if not value:
return {}
ret = json.loads(value)
for (token, props) in ret.items():
checked_props = check_and_sanitize_props(props)
# check token
if token not in log.LOG_CONDITION_TOKEN:
raise ValueError("unsupported log condition token: %r" % token)
# check condition entry
for cond_name in checked_props:
if cond_name not in log.LOG_CONDITION_CONDITION:
raise ValueError("unsupported log condition: %r" % cond_name)
if cond_name == "match":
if log.LOG_CONDITION_TOKEN[token] == "str":
if checked_props[cond_name] not in log.LOG_CONDITION_MATCH_STR:
raise ValueError("unsupported log match: %r" % checked_props[cond_name])
elif log.LOG_CONDITION_TOKEN[token] == "int":
if checked_props[cond_name] not in log.LOG_CONDITION_MATCH_INT:
raise ValueError("unsupported log match: %r" % checked_props[cond_name])
elif log.LOG_CONDITION_TOKEN[token] == "ipaddress":
if checked_props[cond_name] not in log.LOG_CONDITION_MATCH_IPADDRESS + log.LOG_CONDITION_MATCH_IPNETWORK:
raise ValueError("unsupported log match: %r" % checked_props[cond_name])
else:
raise RuntimeError("unsupported log match (fix code): %r" % cond_name)
if cond_name == "value":
if log.LOG_CONDITION_TOKEN[token] == "str":
if checked_props["match"] == "re":
try:
if re.match(checked_props[cond_name], "Test"):
pass
else:
pass
except Exception as e:
raise ValueError("unsupported log match value(re) for condition %r: %r (%s)" % (checked_props["match"], checked_props[cond_name], e))
pass
elif log.LOG_CONDITION_TOKEN[token] == "int":
if str(int(checked_props[cond_name])) != checked_props[cond_name]:
raise ValueError("unsupported log match value(int): %r" % checked_props[cond_name])
if token == "status":
# only 100-599 are valid
if int(checked_props[cond_name]) < 100 or int(checked_props[cond_name]) > 599:
raise ValueError("unsupported log match value(int) not in range 100-599: %r" % checked_props[cond_name])
elif log.LOG_CONDITION_TOKEN[token] == "ipaddress":
try:
if checked_props["match"] in log.LOG_CONDITION_MATCH_IPADDRESS:
ip = ipaddress.ip_address(checked_props[cond_name]) # noqa: F841
elif checked_props["match"] in log.LOG_CONDITION_MATCH_IPNETWORK:
ip_net = ipaddress.ip_network(checked_props[cond_name], strict=True) # noqa: F841
except Exception as e:
raise ValueError("unsupported log match value(ipaddress) for condition %r: %r (%s)" % (checked_props["match"], checked_props[cond_name], e))
# check condition entries are complete
for cond_name in log.LOG_CONDITION_CONDITION:
if cond_name not in checked_props:
raise ValueError("incomplete condition, misses: %r" % cond_name)
# all checks passed
ret[token] = checked_props
return ret
INTERNAL_OPTIONS: Sequence[str] = ("_allow_extra",)
# Default configuration
DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
@@ -667,6 +729,22 @@ This is an automated message. Please do not reply.""",
"value": "False",
"help": "log response content on level=debug",
"type": bool}),
("request_header_on_notice_condition", {
"value": "{}",
"help": "log request header on level=notice with condition",
"type": json_str_condition}),
("request_content_on_notice_condition", {
"value": "{}",
"help": "log request content on level=notice with condition",
"type": json_str_condition}),
("response_header_on_notice_condition", {
"value": "{}",
"help": "log response header on level=notice with condition",
"type": json_str_condition}),
("response_content_on_notice_condition", {
"value": "{}",
"help": "log response content on level=notice with condition",
"type": json_str_condition}),
("rights_rule_doesnt_match_on_debug", {
"value": "False",
"help": "log rights rules which doesn't match on level=debug",
@@ -856,8 +934,7 @@ class Configuration:
except Exception as e:
raise RuntimeError(
"Invalid %s value for option %r in section %r in %s: "
"%r" % (type_.__name__, option, section, source,
raw_value)) from e
"%r (%s)" % (type_.__name__, option, section, source, raw_value, e)) from e
self._configs.append((config, source, bool(privileged)))
for section in new_values:
self._values[section] = self._values.get(section, {})

25
radicale/httputils.py
View File

@@ -32,7 +32,7 @@ import time
from http import client
from typing import List, Mapping, Union, cast
from radicale import config, pathutils, types, utils
from radicale import config, log, pathutils, types, utils
from radicale.log import logger
if sys.version_info < (3, 9):
@@ -151,16 +151,29 @@ def read_raw_request_body(configuration: "config.Configuration",
def read_request_body(configuration: "config.Configuration",
environ: types.WSGIEnviron) -> str:
environ: types.WSGIEnviron, request_info: dict) -> str:
content = decode_request(configuration, environ,
read_raw_request_body(configuration, environ))
if logger.isEnabledFor(logging.DEBUG):
if configuration.get("logging", "request_content_on_debug"):
_limit_content = configuration.get("logging", "limit_content")
_limit_content = configuration.get("logging", "limit_content")
if configuration.get("logging", "request_content_on_debug"):
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content (sha256sum): %s", utils.sha256_str(content))
logger.debug("Request content:\n%s", utils.textwrap_str(content, _limit_content))
else:
_request_content_on_notice_condition = configuration.get("logging", "request_content_on_notice_condition")
if _request_content_on_notice_condition != {}:
if log.log_conditional(
"request-content",
condition=_request_content_on_notice_condition,
value=request_info,
):
logger.notice("Request content (log condition passed):\n%s", utils.textwrap_str(content, _limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content (log condition skipped): suppressed")
else:
logger.debug("Request content: suppressed by config/option [logging] request_content_on_debug")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content: suppressed by config/option [logging] request_content_on_debug")
return content

104
radicale/log.py
View File

@@ -1,7 +1,7 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2011-2017 Guillaume Ayoub
# Copyright © 2017-2023 Unrud <unrud@outlook.com>
# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -28,8 +28,10 @@ Log messages are sent to the first available target of:
import contextlib
import io
import ipaddress
import logging
import os
import re
import socket
import struct
import sys
@@ -48,6 +50,19 @@ LOGGER_FORMATS: Mapping[str, str] = {
}
DATE_FORMAT: str = "%Y-%m-%d %H:%M:%S %z"
LOG_CONDITION_TOKEN: dict = {"method": "str",
"path": "str",
"useragent": "str",
"host": "ipaddress",
"login": "str",
"status": "int",
}
LOG_CONDITION_CONDITION: list = ["match", "value"]
LOG_CONDITION_MATCH_STR: list = ["startswith", "endswith", "equal", "re", "==", "="]
LOG_CONDITION_MATCH_INT: list = ["=", ">", "<", "<=", ">=", "equal", "==", "<>", "!="]
LOG_CONDITION_MATCH_IPADDRESS: list = ["==", "equal", "=", "<>", "!="]
LOG_CONDITION_MATCH_IPNETWORK: list = ["included", "excluded", "incl", "excl"]
LOG_LEVEL_OPTIONS: list = ["trace", "debug", "info", "notice", "warning", "error", "critical", "alert"]
LOG_LEVEL_TRACE: int = 5
@@ -326,3 +341,90 @@ def set_level(level: Union[int, str], backtrace_on_debug: bool, trace_filter: st
logger.addFilter(PassTRACETOKENFilter(trace_filter))
else:
logger.trace("Logging messages on 'trace' level enabled")
def log_conditional(name: str, condition: dict, value: dict) -> bool:
logger.trace("log/conditional/%s/CHECK : condition=%r value=%r", name, condition, value)
# "and" combination
condition_active = False
for token in condition:
condition_active = True
if LOG_CONDITION_TOKEN[token] == "str":
logger.trace("log/conditional/%s/%s/string/CHECK : %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
if condition[token]["match"] in ["equal", "==", "="]:
if not condition[token]["value"] == value[token]:
return False
elif condition[token]["match"] in ["startswith"]:
if not value[token].startswith(condition[token]["value"]):
return False
elif condition[token]["match"] in ["endswith"]:
if not value[token].endswith(condition[token]["value"]):
return False
elif condition[token]["match"] in ["re"]:
if not re.search(condition[token]["value"], value[token]):
return False
logger.trace("log/conditional/%s/%s/string/PASSED : %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
elif LOG_CONDITION_TOKEN[token] == "int":
condition[token]["value"] = int(condition[token]["value"])
logger.trace("log/conditional/%s/%s/integer/CHECK : %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
if condition[token]["match"] in ["equal", "==", "="]:
if not condition[token]["value"] == value[token]:
return False
elif condition[token]["match"] in ["<"]:
if not condition[token]["value"] < value[token]:
return False
elif condition[token]["match"] in ["<="]:
if not condition[token]["value"] <= value[token]:
return False
elif condition[token]["match"] in [">"]:
if not condition[token]["value"] > value[token]:
return False
elif condition[token]["match"] in [">="]:
if not condition[token]["value"] >= value[token]:
return False
elif condition[token]["match"] in ["<>", "!="]:
if not condition[token]["value"] != value[token]:
return False
logger.trace("log/conditional/%s/%s/integer/PASSED : %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
elif LOG_CONDITION_TOKEN[token] == "ipaddress":
logger.trace("log/conditional/%s/%s/ip/CHECK : %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
# simple IP address check
if condition[token]["match"] in ["equal", "==", "="]:
if not condition[token]["value"] == value[token]:
return False
elif condition[token]["match"] in ["<>", "!="]:
if not condition[token]["value"] != value[token]:
return False
elif condition[token]["match"] in ["included", "incl", "excluded", "excl"]:
if condition[token]["match"] in ["included", "incl"]:
if value[token] == "unknown":
return False
else:
ip_net = ipaddress.ip_network(value[token])
ip = ipaddress.ip_network(condition[token]["value"])
if type(ip_net) is ipaddress.IPv4Network and type(ip) is ipaddress.IPv4Network:
if not ip_net.subnet_of(ip):
return False
elif type(ip_net) is ipaddress.IPv6Network and type(ip) is ipaddress.IPv6Network:
if not ip_net.subnet_of(ip):
return False
else:
return False
elif condition[token]["match"] in ["excluded", "excl"]:
if value[token] != "unknown":
ip_net = ipaddress.ip_network(value[token])
if type(ip_net) is ipaddress.IPv4Network and type(ip) is ipaddress.IPv4Network:
if ip_net.subnet_of(ip):
return False
elif type(ip_net) is ipaddress.IPv6Network and type(ip) is ipaddress.IPv6Network:
if ip_net.subnet_of(ip):
return False
logger.trace("log/conditional/%s/%s/ip/PASSED : %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
else:
logger.trace("log/conditional/%s/%s/ip/SKIPPED: %r %s %r", name, token, condition[token]["value"], condition[token]["match"], value[token])
# only return 'True' if at least one condition was found
logger.trace("log/conditional/%s/PASSED: condition=%r value=%r", name, condition, value)
return condition_active

4
radicale/sharing/__init__.py
View File

@@ -511,7 +511,7 @@ class BaseSharing:
return None
# *** POST API ***
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse:
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str, request_info: dict) -> types.WSGIResponse:
# Late import to avoid circular dependency in config
from radicale.app import base as app_base
from radicale.app.base import Access
@@ -616,7 +616,7 @@ class BaseSharing:
logger.trace("sharing/API: called by authenticated user: %r", user)
# read POST data
try:
request_body = httputils.read_request_body(self.configuration, environ)
request_body = httputils.read_request_body(self.configuration, environ, request_info)
except RuntimeError as e:
logger.warning("Bad POST request on %r (read_request_body): %s", path, e, exc_info=True)
return httputils.bad_request("Failed read POST request body")

8
radicale/tests/custom/web.py
View File

@@ -1,6 +1,6 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2021 Unrud <unrud@outlook.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -28,10 +28,10 @@ from radicale import httputils, types, web
class Web(web.BaseWeb):
def get(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
return client.OK, {"Content-Type": "text/plain"}, "custom", None
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
content = httputils.read_request_body(self.configuration, environ)
user: str, request_info: dict) -> types.WSGIResponse:
content = httputils.read_request_body(self.configuration, environ, request_info)
return client.OK, {"Content-Type": "text/plain"}, "echo:" + content, None

230
radicale/tests/test_base.py
View File

@@ -2458,3 +2458,233 @@ permissions: RrWw""")
self.mkcalendar("/calendar.ics/")
event = get_file_content("event_timezone_seconds.ics")
self.put("/calendar.ics/event.ics", event)
def test_logging_conditional_basic(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"request_header_on_notice_condition": '{"method": {"match": "equal", "value": "GET"}, "login": {"match": "equal", "value": "owner"}}',
"request_content_on_notice_condition": '{"method": {"match": "equal", "value": "PUT"}, "useragent": {"match": "startswith", "value": "caldavsync"}}',
"response_header_on_notice_condition": '{"method": {"match": "equal", "value": "PUT"}, "status": {"match": ">=", "value": "201"}, "host": {"match": "included", "value": "127.0.0.0/8"}}',
"response_content_on_notice_condition": '{"method": {"match": "equal", "value": "GET"}, "path": {"match": "endswith", "value": ".ics"}}',
}})
self.mkcalendar("/test/")
event = get_file_content("event1.ics")
path = "/test/event1.ics"
path2 = "/test/event2.ics"
logging.info("\n*** check log condition: Response header (found)")
caplog.clear()
self.put(path, event, remote_host='127.0.0.1')
assert "Response header (log condition passed)" in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response header (not found)")
caplog.clear()
self.put(path, event, remote_host='192.0.2.1', check=204)
assert "Response header (log condition passed)" not in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response content (found)")
caplog.clear()
self.get(path)
assert "Response content (nonXML, log condition passed)" in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response content (not found)")
caplog.clear()
self.get(path2, check=404)
assert "Response content (log condition passed)" not in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Request header (found)")
caplog.clear()
self.get(path2, check=401, login="owner:ownerpw")
assert "Request header (log condition passed)" in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Request content (found)")
caplog.clear()
self.put(path, event, remote_useragent='caldavsync', check=204)
assert "Request content (log condition passed)" in "\n".join(caplog.messages)
def test_logging_conditional_ip_included(self, caplog) -> None:
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_header_on_notice_condition": '{"host": {"match": "included", "value": "127.0.0.0/8"}}',
}})
self.mkcalendar("/test/")
event = get_file_content("event1.ics")
path = "/test/event1.ics"
logging.info("\n*** check log condition: Response header (not found)")
caplog.clear()
self.put(path, event, remote_host='192.0.2.1')
assert "Response header (log condition passed)" not in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response header (found)")
caplog.clear()
self.put(path, event, remote_host='127.0.0.1', check=204)
assert "Response header (log condition passed)" in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response header (not found, no remote_host)")
caplog.clear()
self.put(path, event, check=204)
assert "Response header (log condition passed)" not in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response header (IPv6, not found)")
caplog.clear()
self.put(path, event, check=204, remote_host='2001:db8::1')
assert "Response header (log condition passed)" not in "\n".join(caplog.messages)
def test_logging_conditional_report(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_content_on_notice_condition": '{"status": {"match": "equal", "value": "207"}, "method": {"match": "equal", "value": "REPORT"}}',
}})
self.mkcalendar("/test/")
logging.info("\n*** check log condition: Response content (found)")
caplog.clear()
_, responses = self.report("/test/", """\
<?xml version="1.0" encoding="utf-8" ?>
<C:calendar-query xmlns:C="urn:ietf:params:xml:ns:caldav">
<D:prop xmlns:D="DAV:">
<D:getetag/>
</D:prop>
</C:calendar-query>""")
assert "Response content (XML, log condition passed)" in "\n".join(caplog.messages)
def test_logging_conditional_propfind(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_content_on_notice_condition": '{"status": {"match": "equal", "value": "207"}, "method": {"match": "equal", "value": "PROPFIND"}}',
}})
self.mkcalendar("/test/")
logging.info("\n*** check log condition: Response content (found)")
caplog.clear()
_, responses = self.propfind("/test/")
assert "Response content (XML, log condition passed)" in "\n".join(caplog.messages)
def test_logging_conditional_proppatch(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_content_on_notice_condition": '{"status": {"match": "equal", "value": "207"}, "method": {"match": "equal", "value": "PROPPATCH"}}',
}})
self.mkcalendar("/test/")
proppatch = get_file_content("proppatch_set_calendar_color.xml")
logging.info("\n*** check log condition: Response content (found)")
caplog.clear()
_, responses = self.proppatch("/test/", proppatch)
assert "Response content (XML, log condition passed)" in "\n".join(caplog.messages)
def test_logging_conditional_mkcalendar(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"request_header_on_notice_condition": '{"method": {"match": "equal", "value": "MKCALENDAR"}}',
}})
logging.info("\n*** check log condition: Request content (found)")
caplog.clear()
self.mkcalendar("/test/")
assert "Request header (log condition passed)" in "\n".join(caplog.messages)
def test_logging_conditional_mkcol(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "True",
"request_content_on_notice_condition": '{"method": {"match": "equal", "value": "MKCOL"}}',
}})
logging.info("\n*** check log condition: Request content (found)")
caplog.clear()
self.create_addressbook("/test/")
def test_logging_conditional_regex_basic(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_header_on_notice_condition": '{"method": {"match": "equal", "value": "PUT"}, "path": {"match": "re", "value": "ev?nt[01]"}}',
}})
self.mkcalendar("/test/")
event = get_file_content("event1.ics")
event2 = get_file_content("event2.ics")
path = "/test/event1.ics"
path2 = "/test/event2.ics"
logging.info("\n*** check log condition: Response header (found/re)")
caplog.clear()
self.put(path, event, remote_host='127.0.0.1')
assert "Response header (log condition passed)" in "\n".join(caplog.messages)
logging.info("\n*** check log condition: Response header (found/re)")
caplog.clear()
self.put(path2, event2, remote_host='127.0.0.1')
assert "Response header (log condition passed)" not in "\n".join(caplog.messages)
def test_logging_conditional_problems(self, caplog) -> None:
logging.info("\n*** check log condition config: re ok")
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"request_header_on_notice_condition": '{"path": {"match": "re", "value": "ev?.*nt[01]"}}',
}})
self.mkcalendar("/test1/")
logging.info("\n*** check log condition config: re broken")
try:
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"request_header_on_notice_condition": '{"path": {"match": "re", "value": "*.ics"}}',
}})
except RuntimeError:
pass
else:
raise
logging.info("\n*** check log condition config: status not supported <100")
try:
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_header_on_notice_condition": '{"status": {"match": "<=", "value": "99"}}',
}})
except RuntimeError:
pass
else:
raise
logging.info("\n*** check log condition config: status not supported >599")
try:
self.configure({"logging": {"request_header_on_debug": "False",
"request_content_on_debug": "False",
"response_header_on_debug": "False",
"response_content_on_debug": "False",
"response_header_on_notice_condition": '{"status": {"match": "<=", "value": "600"}}',
}})
except RuntimeError:
pass
else:
raise

11
radicale/web/__init__.py
View File

@@ -1,5 +1,6 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
# Copyright © 2017-2021 Unrud <unrud@outlook.com>
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -49,7 +50,7 @@ class BaseWeb:
self.configuration = configuration
def get(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
"""GET request.
``base_prefix`` is sanitized and never ends with "/".
@@ -58,11 +59,13 @@ class BaseWeb:
``user`` is empty for anonymous users.
``request_info`` dict of additional information
"""
return httputils.METHOD_NOT_ALLOWED
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
"""POST request.
``base_prefix`` is sanitized and never ends with "/".
@@ -71,6 +74,8 @@ class BaseWeb:
``user`` is empty for anonymous users.
``request_info`` dict of additional information
Use ``httputils.read*_request_body(self.configuration, environ)`` to
read the body.

5
radicale/web/internal.py
View File

@@ -1,5 +1,6 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -34,6 +35,6 @@ FALLBACK_MIMETYPE = httputils.FALLBACK_MIMETYPE # deprecated
class Web(web.BaseWeb):
def get(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
return httputils.serve_resource("radicale.web", "internal_data",
base_prefix, path)

4
radicale/web/none.py
View File

@@ -1,6 +1,6 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2025-2025 Peter Bieringer <pb@bieringer.de>
# Copyright © 2025-2026 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -28,7 +28,7 @@ from radicale import httputils, pathutils, types, web
class Web(web.BaseWeb):
def get(self, environ: types.WSGIEnviron, base_prefix: str, path: str,
user: str) -> types.WSGIResponse:
user: str, request_info: dict) -> types.WSGIResponse:
assert path == "/.web" or path.startswith("/.web/")
assert pathutils.sanitize_path(path) == path
if path != "/.web":