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

Merge pull request #2058 from pbiering/sharing-add-auth-test-delay

Browse Source 
Sharing add auth / test delay
This commit is contained in:
Peter Bieringer
2026-03-30 05:37:57 +02:00
committed by GitHub
parent 3ee679e2a8 7fac5e4d7c
commit c0c2efd8c5
6 changed files with 133 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
DOCUMENTATION.md
View File

@@ -253,11 +253,14 @@ The default values should be fine for most scenarios.
max_connections = 20
# 100 Megabyte
max_content_length = 100000000
# 10 Megabyte (>= 3.5.10)
max_resource_size = 10000000
# 30 seconds
timeout = 30
[auth]
# Average delay after failed login attempts in seconds
# Also used for invalid/not-existing/not-enabled share-by-token (>= 3.7.0)
delay = 1
```
@@ -1077,6 +1080,8 @@ Default: `False`
Average delay (in seconds) after failed login attempts.
Also used for invalid/not-existing/not-enabled share-by-token. _(>= 3.7.0)_
Default: `1`
##### realm

3
config
View File

@@ -31,7 +31,7 @@
# Max resource size (bytes), default: 10 Mbyte
# Limited to 80% of max_content_length to cover plain base64 encoded payload
# Announced to clients requesting "max-resource-size" via PROPFIND
#max_ressource_size = 10000000
#max_resource_size = 10000000
# Socket timeout (seconds)
#timeout = 30
@@ -181,6 +181,7 @@
#htpasswd_cache = False
# Incorrect authentication delay (seconds)
# Also used for invalid/not-existing/not-enabled share-by-token (>= 3.7.0)
#delay = 1
# Message displayed in the client when a password is needed

8
radicale/auth/__init__.py
View File

@@ -225,7 +225,7 @@ class BaseAuth:
See also issue 591
"""
time_delta = (time.time_ns() - time_ns_begin) / 1000 / 1000 / 1000
time_delta = (time.time_ns() - time_ns_begin) / 10**9
with self._lock:
# avoid that another thread is changing global value at the same time
failed_auth_delay = self._failed_auth_delay
@@ -268,7 +268,7 @@ class BaseAuth:
cache_failed_cleanup = dict()
for digest in self._cache_failed:
(time_ns_cache, login_cache) = self._cache_failed[digest]
age_failed = int((time_ns - time_ns_cache) / 1000 / 1000 / 1000)
age_failed = int((time_ns - time_ns_cache) / 10**9)
if age_failed > self._cache_failed_logins_expiry:
cache_failed_cleanup[digest] = (login_cache, age_failed)
cache_failed_cleanup_entries = len(cache_failed_cleanup)
@@ -285,7 +285,7 @@ class BaseAuth:
if self._cache_failed.get(digest_failed):
# login+password found in cache "failed" -> shortcut return
(time_ns_cache, login_cache) = self._cache_failed[digest]
age_failed = int((time_ns - time_ns_cache) / 1000 / 1000 / 1000)
age_failed = int((time_ns - time_ns_cache) / 10**9)
logger.debug("Login failed cache entry for user+password found: '%s' (age: %d sec)", login_cache, age_failed)
self._sleep_for_constant_exec_time(time_ns_begin)
return ("", self._type + " / cached")
@@ -294,7 +294,7 @@ class BaseAuth:
(digest_cache, time_ns_cache) = self._cache_successful[login]
digest = self._cache_digest(login, password, str(time_ns_cache))
if digest == digest_cache:
age_success = int((time_ns - time_ns_cache) / 1000 / 1000 / 1000)
age_success = int((time_ns - time_ns_cache) / 10**9)
if age_success > self._cache_successful_logins_expiry:
logger.debug("Login successful cache entry for user+password found but expired: '%s' (age: %d > %d sec)", login, age_success, self._cache_successful_logins_expiry)
# delete expired success from cache

8
radicale/sharing/__init__.py
View File

@@ -19,8 +19,10 @@ 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
@@ -138,6 +140,7 @@ class BaseSharing:
_storage: storage.BaseStorage
_rights: rights.BaseRights
_auth_delay: float
_enabled: bool = False
default_permissions_create_token: str
default_permissions_create_map: str
@@ -154,6 +157,7 @@ class BaseSharing:
self.configuration = configuration
self._rights = rights.load(configuration)
self._storage = storage.load(configuration)
self._auth_delay = configuration.get("auth", "delay")
# Sharing
self.sharing_collection_by_map = configuration.get("sharing", "collection_by_map")
self.sharing_collection_by_token = configuration.get("sharing", "collection_by_token")
@@ -388,6 +392,10 @@ 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):

24
radicale/tests/test_auth.py
View File

@@ -26,6 +26,7 @@ import base64
import logging
import os
import sys
import time
from typing import Iterable, Tuple, Union
import pytest
@@ -60,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") -> None:
= "ascii", delay: int = 0) -> None:
"""Test htpasswd authentication with user "tmp" and password "bepo" for
``test_matrix`` "ascii" or user "😀" and password "🔑" for
``test_matrix`` "unicode"."""
@@ -69,7 +70,7 @@ class TestBaseAuthRequests(BaseTest):
with open(htpasswd_file_path, "w", encoding=encoding) as f:
f.write(htpasswd_content)
self.configure({"auth": {"type": "htpasswd",
"delay": 0,
"delay": delay,
"htpasswd_filename": htpasswd_file_path,
"htpasswd_encryption": htpasswd_encryption},
"server": {"delay_on_error": 0}})
@@ -192,7 +193,7 @@ class TestBaseAuthRequests(BaseTest):
def test_htpasswd_login_cache_successful_plain(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"auth": {"cache_logins": "True"}})
self._test_htpasswd("plain", "tmp:bepo", (("tmp", "bepo", True), ("tmp", "bepo", True)))
self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo", True), ("tmp", "bepo", True)])
htpasswd_found = False
htpasswd_cached_found = False
for line in caplog.messages:
@@ -207,7 +208,7 @@ class TestBaseAuthRequests(BaseTest):
def test_htpasswd_login_cache_failed_plain(self, caplog) -> None:
caplog.set_level(logging.INFO)
self.configure({"auth": {"cache_logins": "True"}})
self._test_htpasswd("plain", "tmp:bepo", (("tmp", "bepo1", False), ("tmp", "bepo1", False)))
self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo1", False), ("tmp", "bepo1", False)])
htpasswd_found = False
htpasswd_cached_found = False
for line in caplog.messages:
@@ -218,6 +219,21 @@ class TestBaseAuthRequests(BaseTest):
if (htpasswd_found is False) or (htpasswd_cached_found is False):
raise ValueError("Logging misses expected log lines")
# login cache failed
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()
self._test_htpasswd("plain", "tmp:bepo", [("tmp", "bepo1", False)], delay=delay)
time_ns_end1 = time.time_ns()
time_ns_begin2 = time.time_ns()
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
# htpasswd file cache
def test_htpasswd_file_cache(self, caplog) -> None:
self.configure({"auth": {"htpasswd_cache": "True"}})

100
radicale/tests/test_sharing.py
View File

@@ -24,6 +24,7 @@ import json
import logging
import os
import re
import time
from typing import Dict, Sequence, Tuple, Union
from radicale import sharing, xmlutils
@@ -747,7 +748,7 @@ class TestSharingApiSanity(BaseTest):
assert "C:supported-calendar-component-set" in response
assert "D:current-user-privilege-set" in response
def test_sharing_api_token_usage(self) -> None:
def test_sharing_api_token_usage_basic(self) -> None:
"""share-by-token API tests - real usage."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
@@ -822,10 +823,10 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self._sharing_api_form("token", "enable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status='success'" in answer
logging.info("\n*** fetch collection using invalid token (without credentials)")
logging.info("\n*** fetch collection using invalid token")
_, headers, answer = self.request("GET", "/.token/v1/invalidtoken/", check=401)
logging.info("\n*** fetch collection using token (without credentials)")
logging.info("\n*** fetch collection using token")
_, headers, answer = self.request("GET", token, check=200)
assert "UID:event" in answer
@@ -834,7 +835,7 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status='success'" in answer
logging.info("\n*** fetch collection using disabled token (without credentials)")
logging.info("\n*** fetch collection using disabled token")
_, headers, answer = self.request("GET", token, check=401)
logging.info("\n*** enable token (form->text)")
@@ -842,7 +843,7 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self._sharing_api_form("token", "enable", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status='success'" in answer
logging.info("\n*** fetch collection using token (without credentials)")
logging.info("\n*** fetch collection using token")
_, headers, answer = self.request("GET", token, check=200)
assert "UID:event" in answer
@@ -865,9 +866,96 @@ class TestSharingApiSanity(BaseTest):
form_array = ["PathOrToken=" + token]
_, 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 (without credentials)")
logging.info("\n*** fetch collection using deleted token")
_, headers, answer = self.request("GET", token, check=401)
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
self.configure({"auth": {"type": "htpasswd",
"delay": delay,
"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"}})
form_array: Sequence[str]
json_dict: dict
path_base = "/owner/calendar.ics/"
logging.info("\n*** prepare")
self.mkcalendar(path_base, login="owner:ownerpw")
event = get_file_content("event1.ics")
path = path_base + "/event1.ics"
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*** create token")
form_array = []
form_array.append("PathMapped=" + path_base)
form_array.append("Enabled=True")
_, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status='success'" in answer
assert "PathOrToken=" in answer
# extract token
match = re.search("PathOrToken='(.+)'", answer)
if match:
token = match.group(1)
logging.info("received token %r", token)
else:
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
logging.info("\n*** fetch collection using token")
time_ns_begin = time.time_ns()
_, headers, answer = self.request("GET", token, check=200)
time_ns_end = time.time_ns()
assert (time_ns_end - time_ns_begin) < delay_ns
assert "UID:event" in answer
logging.info("\n*** disable token (form->text)")
form_array = ["PathOrToken=" + token]
_, headers, answer = self._sharing_api_form("token", "disable", check=200, login="owner:ownerpw", form_array=form_array)
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
logging.info("\n*** delete token (json->json)")
json_dict = {'PathOrToken': token}
_, headers, answer = self._sharing_api_json("token", "delete", check=200, login="owner:ownerpw", json_dict=json_dict)
answer_dict = json.loads(answer)
assert answer_dict['ApiVersion'] == 1
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
def test_sharing_api_map_basic(self) -> None:
"""share-by-map API basic tests."""
self.configure({"auth": {"type": "htpasswd",