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

Merge pull request #2063 from pbiering/align-delay

Browse Source 
Align delay on missing auth or server error
This commit is contained in:
Peter Bieringer
2026-04-02 17:31:08 +02:00
committed by GitHub
parent 113fe4fcd7 49db75dae8
commit 68a8c2e44f
9 changed files with 149 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/test.yml vendored
View File

@@ -223,7 +223,7 @@ jobs:
integ-test:
timeout-minutes: 60
runs-on: ubuntu-latest
needs: test-ubuntu-python-newest
needs: [test-ubuntu-python-newest, htmlvalidator, js-test]
steps:
- uses: actions/checkout@v5
- uses: actions/setup-python@v6

8
DOCUMENTATION.md
View File

@@ -838,9 +838,9 @@ Default: `8`
_(>= 3.7.0)_
Base delay in case of error response (seconds)
Base delay in case of error 5xx response (seconds)
Default: `0.01`
Default: `1`
##### max_content_length
@@ -1078,9 +1078,7 @@ Default: `False`
##### delay
Average delay (in seconds) after failed login attempts.
Also used for invalid/not-existing/not-enabled share-by-token. _(>= 3.7.0)_
Average delay (in seconds) after failed or missing login attempts or denied access.
Default: `1`

7
config
View File

@@ -21,8 +21,8 @@
# Max parallel connections
#max_connections = 8
# Base delay in case of error response (seconds)
#delay_on_error = 0.01
# Base delay in case of error 5xx response (seconds)
#delay_on_error = 1
# Max size of request body (bytes), default: 100 Mbyte
# In case of using a reverse proxy in front of check also there related option
@@ -180,8 +180,7 @@
# Enable caching of htpasswd file based on size and mtime_ns
#htpasswd_cache = False
# Incorrect authentication delay (seconds)
# Also used for invalid/not-existing/not-enabled share-by-token (>= 3.7.0)
# Incorrect or missing authentication or denied access average delay (seconds)
#delay = 1
# Message displayed in the client when a password is needed

50
radicale/app/__init__.py
View File

@@ -31,6 +31,7 @@ import cProfile
import datetime
import io
import logging
import os
import pprint
import pstats
import random
@@ -292,6 +293,26 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
logger.debug("Response header: suppressed by config/option [logging] response_header_on_debug")
# Start response
# delay on error
delay: float = 0.0
if (status == 401 or status == 403) and (not login or user):
# delay for required but missing authentication or denied access
if self._auth_delay > 0:
if 'PYTEST_VERSION' in os.environ:
# no random during tests
delay = self._auth_delay
else:
delay = self._auth_delay * (0.5 + random.random())
if status >= 500 and status <= 599:
if self._delay_on_error > 0:
delay = self._delay_on_error
if delay > 0:
if logger.isEnabledFor(logging.DEBUG):
if 'PYTEST_VERSION' in os.environ:
logger.debug("Response fixed(pytest) delay triggered by result code: %d -> %0.3f seconds", status, delay)
else:
logger.debug("Response random delay triggered by result code: %d -> %0.3f seconds", status, delay)
time.sleep(delay)
time_end = datetime.datetime.now()
time_delta_seconds = (time_end - time_begin).total_seconds()
status_text = "%d %s" % (
@@ -310,19 +331,10 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
flags_text = " (" + " ".join(flags) + ")"
else:
flags_text = ""
# delay on error
if status >= 400:
if self._delay_on_error > 0:
random_delay = self._delay_on_error * (1 + random.random())
if status >= 500:
random_delay = 2 * random_delay
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response delay triggered by result code: %d -> %0.3f seconds", status, random_delay)
time.sleep(random_delay)
if answer is not None:
logger.info("%s response status for %r%s in %.3f seconds %s %s bytes%s: %s",
request_method, unsafe_path, depthinfo,
(time_end - time_begin).total_seconds(), content_encoding, str(len(answer)),
time_delta_seconds, content_encoding, str(len(answer)),
flags_text,
status_text)
else:
@@ -487,8 +499,15 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
remote_host, login, info)
# Random delay to avoid timing oracles and bruteforce attacks
if self._auth_delay > 0:
random_delay = self._auth_delay * (0.5 + random.random())
logger.debug("Failed login, sleeping random: %.3f sec", random_delay)
if 'PYTEST_VERSION' in os.environ:
# no random during tests
random_delay = self._auth_delay
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Failed login, sleeping fixed(pytest): %.3f sec", random_delay)
else:
random_delay = self._auth_delay * (0.5 + random.random())
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Failed login, sleeping random: %.3f sec", random_delay)
time.sleep(random_delay)
if user and not pathutils.is_safe_path_component(user):
@@ -566,12 +585,15 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
self.profiler_per_request_method[request_method].disable()
if (status, headers, answer, xml_request) == httputils.NOT_ALLOWED:
logger.info("Access to %r denied for %s", path,
repr(user) if user else "anonymous user")
if path.startswith("/.token"):
logger.info("Access to %r denied", path)
else:
logger.info("Access to %r denied for %s", path, repr(user) if user else "anonymous user")
else:
status, headers, answer, xml_request = httputils.NOT_ALLOWED
if ((status, headers, answer, xml_request) == httputils.NOT_ALLOWED and not user and
not path.startswith("/.token") and
not external_login):
# Unknown or unauthorized user
logger.debug("Asking client for authentication")

4
radicale/config.py
View File

@@ -172,8 +172,8 @@ DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
"help": "maximum size of request body in bytes (default: 100 Mbyte)",
"type": positive_int}),
("delay_on_error", {
"value": "0.01",
"help": "base delay in case of error response (seconds)",
"value": "1",
"help": "base delay in case of error 5xx response (seconds)",
"type": positive_float}),
("max_resource_size", {
"value": "10000000",

6
radicale/sharing/__init__.py
View File

@@ -19,10 +19,8 @@ import base64
import io
import json
import logging
import random
import re
import socket
import time
import uuid
from csv import DictWriter
from datetime import datetime
@@ -394,10 +392,6 @@ class BaseSharing:
if share is None:
share = self.sharing_collection_by_token_resolver(path)
if share is not None and 'error' in share:
if self._auth_delay > 0:
random_delay = self._auth_delay * (0.5 + random.random())
logger.debug("Failed shared-by-token resolver, sleeping random: %.3f sec", random_delay)
time.sleep(random_delay)
return None
else:
if logger.isEnabledFor(logging.DEBUG):

29
radicale/tests/test_auth.py
View File

@@ -23,10 +23,10 @@ Radicale tests with simple requests and authentication.
"""
import base64
import datetime
import logging
import os
import sys
import time
from typing import Iterable, Tuple, Union
import pytest
@@ -61,7 +61,7 @@ class TestBaseAuthRequests(BaseTest):
def _test_htpasswd(self, htpasswd_encryption: str, htpasswd_content: str,
test_matrix: Union[str, Iterable[Tuple[str, str, bool]]]
= "ascii", delay: int = 0) -> None:
= "ascii", delay: float = 0) -> None:
"""Test htpasswd authentication with user "tmp" and password "bepo" for
``test_matrix`` "ascii" or user "😀" and password "🔑" for
``test_matrix`` "unicode"."""
@@ -223,16 +223,25 @@ class TestBaseAuthRequests(BaseTest):
def test_htpasswd_login_cache_failed_delay_plain(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"auth": {"cache_logins": "True"}})
delay = 1
delay_ns = delay * 10**9 * 0.5 # delay minimum jitter
time_ns_begin1 = time.time_ns()
delay = .3
delay_min = delay * 0.9 # no random jitter during test
delay_max = delay + 0.2 # no random jitter during test
if sys.platform == "darwin": # no reliable sleep times
delay_max = delay_max * 1.5
time_begin = datetime.datetime.now()
self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo1", False)], delay=delay)
time_ns_end1 = time.time_ns()
time_ns_begin2 = time.time_ns()
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
time_begin = datetime.datetime.now()
self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo1", False)], delay=delay)
time_ns_end2 = time.time_ns()
assert (time_ns_end1 - time_ns_begin1) > delay_ns
assert (time_ns_end2 - time_ns_begin2) > delay_ns
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
# htpasswd file cache
def test_htpasswd_file_cache(self, caplog) -> None:

21
radicale/tests/test_base.py
View File

@@ -21,9 +21,11 @@ Radicale tests with simple requests.
"""
import datetime
import logging
import os
import posixpath
import sys
import urllib
from typing import Any, Callable, ClassVar, Iterable, List, Optional, Tuple
@@ -727,7 +729,7 @@ permissions: RrWw""")
assert responses["/calendar.ics/"] == 200
self.get("/calendar.ics/", check=404)
def test_delete_collection_global_forbid(self) -> None:
def test_delete_collection_global_forbid_base(self) -> None:
"""Delete a collection (expect forbidden)."""
self.configure({"rights": {"permit_delete_collection": False}})
self.mkcalendar("/calendar.ics/")
@@ -736,6 +738,23 @@ permissions: RrWw""")
_, responses = self.delete("/calendar.ics/", check=401)
self.get("/calendar.ics/", check=200)
def test_delete_collection_global_forbid_delay(self) -> None:
"""Delete a collection (expect forbidden, check delay)."""
delay = .3
delay_min = delay * 0.9 # no random jitter during test
delay_max = delay + 0.2 # no random jitter during test
if sys.platform == "darwin": # no reliable sleep times
delay_max = delay_max * 1.5
self.configure({"rights": {"permit_delete_collection": False}, "auth": {"delay": delay}})
self.mkcalendar("/calendar.ics/")
time_begin = datetime.datetime.now()
_, responses = self.delete("/calendar.ics/", check=401)
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
def test_delete_collection_global_forbid_explicit_permit(self) -> None:
"""Delete a collection with permitted path (expect permit)."""
self.configure({"rights": {"permit_delete_collection": False}})

86
radicale/tests/test_sharing.py
View File

@@ -20,11 +20,12 @@ Radicale tests related to sharing.
"""
import datetime
import json
import logging
import os
import re
import time
import sys
from typing import Dict, Sequence, Tuple, Union
from radicale import sharing, xmlutils
@@ -202,7 +203,7 @@ class TestSharingApiSanity(BaseTest):
"collection_by_token": "False"}
})
def test_sharing_api_base_no_auth(self) -> None:
def test_sharing_api_base_no_auth_basic(self) -> None:
"""POST request at '/.sharing' without authentication."""
# disabled
for path in ["/.sharing", "/.sharing/"]:
@@ -258,6 +259,39 @@ class TestSharingApiSanity(BaseTest):
})
_, headers, _ = self.request("POST", path, check=401)
def test_sharing_api_base_no_auth_delay(self) -> None:
delay = .3
delay_min = delay * 0.9 # no random jitter during test
delay_max = delay + 0.2 # no random jitter during test
if sys.platform == "darwin": # no reliable sleep times
delay_max = delay_max * 1.5
for path in ["/.sharing", "/.sharing/"]:
time_begin = datetime.datetime.now()
_, headers, _ = self.request("POST", path, check=404)
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta < delay_min # 404 should have no delay
path = "/.sharing/"
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}})
# no database is active
logging.info("\n*** check API hook base: map=True token=False (incl. delay)")
self.configure({"sharing": {
"collection_by_map": "True",
"collection_by_token": "False"},
"auth": {"delay": delay}})
time_begin = datetime.datetime.now()
_, headers, _ = self.request("POST", path, check=401)
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
def test_sharing_api_base_with_auth(self) -> None:
"""POST request at '/.sharing' with authentication."""
self.configure({"auth": {"type": "htpasswd",
@@ -834,7 +868,7 @@ class TestSharingApiSanity(BaseTest):
assert "Status='success'" in answer
logging.info("\n*** fetch collection using invalid token")
_, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=401)
_, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=403)
logging.info("\n*** fetch collection using token")
_, headers, answer = self.request("GET", token, check=200)
@@ -846,7 +880,7 @@ class TestSharingApiSanity(BaseTest):
assert "Status='success'" in answer
logging.info("\n*** fetch collection using disabled token")
_, headers, answer = self.request("GET", token, check=401)
_, headers, answer = self.request("GET", token, check=403)
logging.info("\n*** enable token (form->text)")
form_array = ["PathOrToken=" + token]
@@ -877,12 +911,15 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self._sharing_api_form("token", "delete", check=404, login="owner:ownerpw", form_array=form_array)
logging.info("\n*** fetch collection using deleted token")
_, headers, answer = self.request("GET", token, check=401)
_, headers, answer = self.request("GET", token, check=403)
def test_sharing_api_token_usage_delay(self) -> None:
"""share-by-token API tests - real usage."""
delay = .3
delay_ns = delay * 10**9 * 0.5 # delay minimum jitter
delay_min = delay * 0.9 # no random jitter during test
delay_max = delay + 0.2 # no random jitter during test
if sys.platform == "darwin": # no reliable sleep times
delay_max = delay_max * 1.5
self.configure({"auth": {"type": "htpasswd",
"delay": delay,
@@ -930,16 +967,19 @@ class TestSharingApiSanity(BaseTest):
assert False
logging.info("\n*** fetch collection using invalid token")
time_ns_begin = time.time_ns()
_, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=401)
time_ns_end = time.time_ns()
assert (time_ns_end - time_ns_begin) > delay_ns
time_begin = datetime.datetime.now()
_, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=403)
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
logging.info("\n*** fetch collection using token")
time_ns_begin = time.time_ns()
time_begin = datetime.datetime.now()
_, headers, answer = self.request("GET", token, check=200)
time_ns_end = time.time_ns()
assert (time_ns_end - time_ns_begin) < delay_ns
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta < delay_min # no delay
assert "UID:event" in answer
logging.info("\n*** disable token (form->text)")
@@ -948,10 +988,12 @@ class TestSharingApiSanity(BaseTest):
assert "Status='success'" in answer
logging.info("\n*** fetch collection using disabled token")
time_ns_begin = time.time_ns()
_, headers, answer = self.request("GET", token, check=401)
time_ns_end = time.time_ns()
assert (time_ns_end - time_ns_begin) > delay_ns
time_begin = datetime.datetime.now()
_, headers, answer = self.request("GET", token, check=403)
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
logging.info("\n*** delete token (json->json)")
json_dict = {'PathOrToken': token}
@@ -961,10 +1003,12 @@ class TestSharingApiSanity(BaseTest):
assert answer_dict['Status'] == "success"
logging.info("\n*** fetch collection using deleted token with delay")
time_ns_begin = time.time_ns()
_, headers, answer = self.request("GET", token, check=401)
time_ns_end = time.time_ns()
assert (time_ns_end - time_ns_begin) > delay_ns
time_begin = datetime.datetime.now()
_, headers, answer = self.request("GET", token, check=403)
time_end = datetime.datetime.now()
time_delta = (time_end - time_begin).total_seconds()
assert time_delta > delay_min
assert time_delta < delay_max
def test_sharing_api_map_basic(self) -> None:
"""share-by-map API basic tests."""