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

Merge pull request #1946 from pbiering/add-item-verification-hexdump-checksum

Browse Source 
Minor extensions
This commit is contained in:
Peter Bieringer
2025-12-28 13:46:19 +01:00
committed by GitHub
parent 99ee86c6a4 17a3816e91
commit a7350af005
8 changed files with 220 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@@ -3,6 +3,9 @@
## 3.5.11.dev
* Extend: logwatch script
* Extend: [logging] bad_put_request_content: log checksum and hexdump of request on debug level
* Extend: [logging] request_content_on_debug: log checksum of request on debug level
* Extend: add command line option "--verify-item <file>" for dedicated item file analysis
## 3.5.10
* Improve: logging of broken calendar items during PUT

38
DOCUMENTATION.md
View File

@@ -714,6 +714,44 @@ Reason for problems can be
## Documentation
### Options
#### General Options
##### --version
Print version
##### --verify-storage
Verification of local collections storage
##### --verify-item <file>
_(>= 3.5.11)_
Verification of a particular item file
##### -C|--config <file>
Load one or more specified config file(s)
##### -D|--debug
Turns log level to debug
#### Configuration Options
Each supported option from config file can be provided/overridden by command line
replacing `_` with `-` and prepending the section followed by a `-`, e.g.
```
[logging]
backtrace_on_debug = False
```
can be enabled using `--logging-backtrace-on-debug=true` on command line.
### Configuration
Radicale can be configured with a configuration file or with

19
radicale/__main__.py
View File

@@ -1,7 +1,7 @@
# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2011-2017 Guillaume Ayoub
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
# Copyright © 2024-2025 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
@@ -33,7 +33,7 @@ import sys
from types import FrameType
from typing import List, Optional, cast
from radicale import VERSION, config, log, server, storage, types
from radicale import VERSION, config, item, log, server, storage, types
from radicale.log import logger
@@ -65,6 +65,8 @@ def run() -> None:
parser.add_argument("--version", action="version", version=VERSION)
parser.add_argument("--verify-storage", action="store_true",
help="check the storage for errors and exit")
parser.add_argument("--verify-item", action="store", nargs=1,
help="check the provided item file for errors and exit")
parser.add_argument("-C", "--config",
help="use specific configuration files", nargs="*")
parser.add_argument("-D", "--debug", action="store_const", const="debug",
@@ -194,6 +196,19 @@ def run() -> None:
sys.exit(1)
return
if args_ns.verify_item:
encoding = configuration.get("encoding", "stock")
logger.info("Item verification start using 'stock' encoding: %s", encoding)
try:
if not item.verify(args_ns.verify_item[0], encoding):
logger.critical("Item verification failed")
sys.exit(1)
except Exception as e:
logger.critical("An exception occurred during item "
"verification: %s", e, exc_info=False)
sys.exit(1)
return
# Create a socket pair to notify the server of program shutdown
shutdown_socket, shutdown_socket_out = socket.socketpair()

13
radicale/app/__init__.py
View File

@@ -30,6 +30,7 @@ import base64
import cProfile
import datetime
import io
import logging
import pprint
import pstats
import random
@@ -253,9 +254,11 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
if answer is not None:
if isinstance(answer, str):
if self._response_content_on_debug:
logger.debug("Response content (nonXML):\n%s", utils.textwrap_str(answer))
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content (nonXML):\n%s", utils.textwrap_str(answer))
else:
logger.debug("Response content: suppressed by config/option [logging] response_content_on_debug")
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 = [
@@ -276,9 +279,11 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
headers.update(self._extra_headers)
if self._response_header_on_debug:
logger.debug("Response header:\n%s", utils.textwrap_str(pprint.pformat(headers)))
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response header:\n%s", utils.textwrap_str(pprint.pformat(headers)))
else:
logger.debug("Response header: suppressed by config/option [logging] response_header_on_debug")
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response header: suppressed by config/option [logging] response_header_on_debug")
# Start response
time_end = datetime.datetime.now()

3
radicale/app/put.py
View File

@@ -202,6 +202,9 @@ class ApplicationPartPut(ApplicationBase):
"Bad PUT request on %r (read_components): %s", path, e, exc_info=True)
if self._log_bad_put_request_content:
logger.warning("Bad PUT request content of %r:\n%s", path, utils.textwrap_str(content))
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Request content (sha256sum): %s", utils.sha256_str(content))
logger.debug("Request content (hexdump/lines):\n%s", utils.hexdump_lines(content))
else:
logger.debug("Bad PUT request content: suppressed by config/option [logging] bad_put_request_content")
return httputils.BAD_REQUEST

8
radicale/httputils.py
View File

@@ -24,6 +24,7 @@ Helper functions for HTTP.
"""
import contextlib
import logging
import os
import pathlib
import sys
@@ -150,9 +151,12 @@ def read_request_body(configuration: "config.Configuration",
content = decode_request(configuration, environ,
read_raw_request_body(configuration, environ))
if configuration.get("logging", "request_content_on_debug"):
logger.debug("Request content:\n%s", utils.textwrap_str(content))
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))
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

24
radicale/item/__init__.py
View File

@@ -3,7 +3,8 @@
# Copyright © 2008 Pascal Halter
# Copyright © 2014 Jean-Marc Martins
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2018 Unrud <unrud@outlook.com>
# Copyright © 2017-2022 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 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
@@ -37,7 +38,7 @@ from typing import (Any, Callable, List, MutableMapping, Optional, Sequence,
import vobject
from radicale import storage # noqa:F401
from radicale import pathutils
from radicale import pathutils, utils
from radicale.item import filter as radicale_filter
from radicale.log import logger
@@ -335,6 +336,25 @@ def find_time_range(vobject_item: vobject.base.Component, tag: str
return math.floor(start.timestamp()), math.ceil(end.timestamp())
def verify(file: str, encoding: str):
logger.info("Verifying item: %s", file)
with open(file, "rb") as f:
content_raw = f.read()
content = content_raw.decode(encoding)
logger.info("Verifying item: %s has sha256sum %r", file, utils.sha256_bytes(content_raw))
try:
vobject_items = read_components(content) # noqa: F841
except Exception as e:
logger.error("Verifying item: %s problem: %s", file, e)
logger.warning("Item content:\n%s", utils.textwrap_str(content))
logger.info("Item content (hexdump):\n%s", utils.hexdump_str(content))
logger.info("Item content (hexdump/lines):\n%s", utils.hexdump_lines(content))
return False
else:
logger.info("Verifying item: %s successful", file)
return True
class Item:
"""Class for address book and calendar entries."""

122
radicale/utils.py
View File

@@ -22,7 +22,9 @@ import os
import ssl
import sys
import textwrap
from hashlib import sha256
from importlib import import_module, metadata
from string import ascii_letters, digits, punctuation
from typing import Callable, Sequence, Tuple, Type, TypeVar, Union
from radicale import config
@@ -342,3 +344,123 @@ def limit_str(content: str, limit: int) -> str:
def textwrap_str(content: str, limit: int = 2000) -> str:
# TODO: add support for config option and prefix
return textwrap.indent(limit_str(content, limit), " ", lambda line: True)
def dataToHex(data, count):
result = ''
for item in range(count):
if ((item > 0) and ((item % 8) == 0)):
result += ' '
if (item < len(data)):
result += '%02x' % data[item] + ' '
else:
result += ' '
return result
def dataToAscii(data, count):
result = ''
for item in range(count):
if (item < len(data)):
char = chr(data[item])
if char in ascii_letters or \
char in digits or \
char in punctuation or \
char == ' ':
result += char
else:
result += '.'
return result
def dataToSpecial(data, count):
result = ''
for item in range(count):
if (item < len(data)):
char = chr(data[item])
if char == '\r':
result += 'C'
elif char == '\n':
result += 'L'
elif (ord(char) & 0xf8) == 0xf0: # assuming UTF-8
result += '4'
elif (ord(char) & 0xf0) == 0xf0: # assuming UTF-8
result += '3'
elif (ord(char) & 0xe0) == 0xe0: # assuming UTF-8
result += '2'
else:
result += '.'
return result
def hexdump_str(content: str, limit: int = 2000) -> str:
result = "Hexdump of string: index <bytes> | <ASCII> | <CTRL: C=CR L=LF 2/3/4=UTF-8-length> |\n"
index = 0
size = 16
bytestring = content.encode("utf-8") # assuming UTF-8
length = len(bytestring)
while (index < length) and (index < limit):
data = bytestring[index:index+size]
hex = dataToHex(data, size)
ascii = dataToAscii(data, size)
special = dataToSpecial(data, size)
result += '%08x ' % index
result += hex
result += '|'
result += '%-16s' % ascii
result += '|'
result += '%-16s' % special
result += '|'
result += '\n'
index += size
return result
def hexdump_line(line: str, limit: int = 200) -> str:
result = ""
length_str = len(line)
bytestring = line.encode("utf-8") # assuming UTF-8
length = len(bytestring)
size = length
if (size > limit):
size = limit
hex = dataToHex(bytestring, size)
ascii = dataToAscii(bytestring, size)
special = dataToSpecial(bytestring, size)
result += '%3d/%3d' % (length_str, length)
result += ': '
result += hex
result += '|'
result += ascii
result += '|'
result += special
result += '|'
result += '\n'
return result
def hexdump_lines(lines: str, limit: int = 200) -> str:
result = "Hexdump of lines: nr chars/bytes: <bytes> | <ASCII> | <CTRL: C=CR L=LF 2/3/4=UTF-8-length> |\n"
counter = 0
for line in lines.splitlines(True):
result += '% 4d ' % counter
result += hexdump_line(line)
counter += 1
return result
def sha256_str(content: str) -> str:
_hash = sha256()
_hash.update(content.encode("utf-8")) # assuming UTF-8
return _hash.hexdigest()
def sha256_bytes(content: bytes) -> str:
_hash = sha256()
_hash.update(content)
return _hash.hexdigest()