studio/calendars
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🏗️(caldav) migrate from davical to sabre/dav

Browse Source
This commit is contained in:
Sylvain Zimmer
2026-01-11 02:28:04 +01:00
parent a36348ead1
commit bc801d3007
35 changed files with 1059 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/backend/core/tests/authentication/test_backends.py
View File

@@ -500,7 +500,7 @@ def test_authentication_session_tokens(
status=200,
)
with django_assert_num_queries(27):
with django_assert_num_queries(7):
user = klass.authenticate(
request,
code="test-code",

25
src/backend/core/tests/conftest.py
View File

@@ -19,23 +19,32 @@ VIA = [USER, TEAM]
@pytest.fixture(autouse=True)
def truncate_davical_tables(django_db_setup, django_db_blocker):
"""Fixture to truncate DAViCal tables at the start of each test.
def truncate_caldav_tables(django_db_setup, django_db_blocker):
"""Fixture to truncate CalDAV server tables at the start of each test.
DAViCal tables are created by the DAViCal container migrations, not Django.
CalDAV server tables are created by the CalDAV server container migrations, not Django.
We just truncate them to ensure clean state for each test.
"""
with django_db_blocker.unblock():
with connection.cursor() as cursor:
# Truncate DAViCal tables if they exist (created by DAViCal container)
# Truncate CalDAV server tables if they exist (created by CalDAV server container)
cursor.execute("""
DO $$
BEGIN
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'principal') THEN
TRUNCATE TABLE principal CASCADE;
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'principals') THEN
TRUNCATE TABLE principals CASCADE;
END IF;
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'usr') THEN
TRUNCATE TABLE usr CASCADE;
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'users') THEN
TRUNCATE TABLE users CASCADE;
END IF;
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'calendars') THEN
TRUNCATE TABLE calendars CASCADE;
END IF;
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'calendarinstances') THEN
TRUNCATE TABLE calendarinstances CASCADE;
END IF;
IF EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'calendarobjects') THEN
TRUNCATE TABLE calendarobjects CASCADE;
END IF;
END $$;
""")

5
src/backend/core/tests/test_api_users.py
View File

@@ -148,11 +148,14 @@ def test_api_users_list_throttling_authenticated(settings):
assert response.status_code == 429
def test_api_users_list_query_email():
def test_api_users_list_query_email(settings):
"""
Authenticated users should be able to list users and filter by email.
Only exact email matches are returned (case-insensitive).
"""
settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"]["user_list_burst"] = "9999/minute"
user = factories.UserFactory()
client = APIClient()

307
src/backend/core/tests/test_caldav_proxy.py Normal file
View File

@@ -0,0 +1,307 @@
"""Tests for CalDAV proxy view."""
from xml.etree import ElementTree as ET
from django.conf import settings
import pytest
import responses
from rest_framework.status import (
HTTP_200_OK,
HTTP_207_MULTI_STATUS,
HTTP_401_UNAUTHORIZED,
)
from rest_framework.test import APIClient
from core import factories
@pytest.mark.django_db
class TestCalDAVProxy:
"""Tests for CalDAVProxyView."""
def test_proxy_requires_authentication(self):
"""Test that unauthenticated requests return 401."""
client = APIClient()
response = client.generic("PROPFIND", "/api/v1.0/caldav/")
assert response.status_code == HTTP_401_UNAUTHORIZED
@responses.activate
def test_proxy_forwards_headers_correctly(self):
"""Test that proxy forwards X-Forwarded-User headers."""
user = factories.UserFactory(email="test@example.com")
client = APIClient()
client.force_login(user)
# Mock CalDAV server response
caldav_url = settings.CALDAV_URL
responses.add(
responses.Response(
method="PROPFIND",
url=f"{caldav_url}/",
status=HTTP_207_MULTI_STATUS,
body='<?xml version="1.0"?><multistatus xmlns="DAV:"></multistatus>',
headers={"Content-Type": "application/xml"},
)
)
response = client.generic("PROPFIND", "/api/v1.0/caldav/")
# Verify request was made to CalDAV server
assert len(responses.calls) == 1
request = responses.calls[0].request
# Verify headers were forwarded
assert request.headers["X-Forwarded-User"] == user.email
assert request.headers["X-Forwarded-Host"] is not None
assert request.headers["X-Forwarded-Proto"] == "http"
@responses.activate
def test_proxy_ignores_client_sent_x_forwarded_user_header(self):
"""Test that proxy ignores and overwrites any X-Forwarded-User header sent by client.
This is a security test to ensure that hostile clients cannot impersonate other users
by sending a malicious X-Forwarded-User header. The proxy should always use the
authenticated Django user's email, not any header value sent by the client.
"""
user = factories.UserFactory(email="legitimate@example.com")
client = APIClient()
client.force_login(user)
# Mock CalDAV server response
caldav_url = settings.CALDAV_URL
responses.add(
responses.Response(
method="PROPFIND",
url=f"{caldav_url}/api/v1.0/caldav/",
status=HTTP_207_MULTI_STATUS,
body='<?xml version="1.0"?><multistatus xmlns="DAV:"></multistatus>',
headers={"Content-Type": "application/xml"},
)
)
# Try to send a malicious X-Forwarded-User header as if we were another user
malicious_email = "attacker@example.com"
response = client.generic(
"PROPFIND",
"/api/v1.0/caldav/",
HTTP_X_FORWARDED_USER=malicious_email,
)
# Verify request was made to CalDAV server
assert len(responses.calls) == 1
request = responses.calls[0].request
# Verify that the X-Forwarded-User header uses the authenticated user's email,
# NOT the malicious header value sent by the client
assert request.headers["X-Forwarded-User"] == user.email, (
f"Expected X-Forwarded-User to be {user.email} (authenticated user), "
f"but got {request.headers.get('X-Forwarded-User')}. "
f"This indicates a security vulnerability - client-sent headers are being trusted!"
)
assert request.headers["X-Forwarded-User"] != malicious_email, (
"X-Forwarded-User should NOT use client-sent header value"
)
@pytest.mark.skipif(
not settings.CALDAV_URL,
reason="CalDAV server URL not configured - integration test requires real server",
)
def test_proxy_propfind_response_contains_prefixed_urls(self):
"""Integration test: PROPFIND responses from sabre/dav should contain URLs with proxy prefix.
This test verifies that sabre/dav's BaseUriPlugin correctly uses X-Forwarded-Prefix
to generate URLs with the proxy prefix. It requires the CalDAV server to be running.
Note: This test does NOT use @responses.activate as it needs to hit the real server.
"""
user = factories.UserFactory(email="test@example.com")
client = APIClient()
client.force_login(user)
# Make actual request to CalDAV server through proxy
# The server should use X-Forwarded-Prefix to generate URLs
propfind_body = '<?xml version="1.0"?><propfind xmlns="DAV:"><prop><resourcetype/></prop></propfind>'
response = client.generic(
"PROPFIND",
"/api/v1.0/caldav/",
data=propfind_body,
content_type="application/xml",
)
assert response.status_code == HTTP_207_MULTI_STATUS, (
f"Expected 207 Multi-Status, got {response.status_code}: {response.content.decode('utf-8', errors='ignore')}"
)
# Parse the response XML
root = ET.fromstring(response.content)
# Find all href elements
href_elems = root.findall(".//{DAV:}href")
assert len(href_elems) > 0, "PROPFIND response should contain href elements"
# Verify all URLs that start with /principals/ or /calendars/ include the proxy prefix
# This verifies that sabre/dav's BaseUriPlugin is working correctly
for href_elem in href_elems:
href = href_elem.text
if href and (
href.startswith("/principals/") or href.startswith("/calendars/")
):
assert href.startswith("/api/v1.0/caldav/"), (
f"Expected URL to start with /api/v1.0/caldav/, got {href}. "
f"This indicates sabre/dav BaseUriPlugin is not using X-Forwarded-Prefix correctly. "
f"Full response: {response.content.decode('utf-8', errors='ignore')}"
)
@responses.activate
def test_proxy_passes_through_calendar_urls(self):
"""Test that calendar URLs in PROPFIND responses are passed through unchanged.
Since we removed URL rewriting from the proxy, sabre/dav should generate
URLs with the correct prefix. This test verifies the proxy passes responses through.
"""
user = factories.UserFactory(email="test@example.com")
client = APIClient()
client.force_login(user)
# Mock CalDAV server PROPFIND response with calendar URL that already has prefix
# (sabre/dav should generate URLs with prefix when X-Forwarded-Prefix is set)
caldav_url = settings.CALDAV_URL
propfind_xml = """<?xml version="1.0"?>
<multistatus xmlns="DAV:">
<response>
<href>/api/v1.0/caldav/calendars/test@example.com/calendar-id/</href>
<propstat>
<prop>
<resourcetype>
<collection/>
<calendar xmlns="urn:ietf:params:xml:ns:caldav"/>
</resourcetype>
</prop>
</propstat>
</response>
</multistatus>"""
responses.add(
responses.Response(
method="PROPFIND",
url=f"{caldav_url}/api/v1.0/caldav/",
status=HTTP_207_MULTI_STATUS,
body=propfind_xml,
headers={"Content-Type": "application/xml"},
)
)
response = client.generic("PROPFIND", "/api/v1.0/caldav/")
assert response.status_code == HTTP_207_MULTI_STATUS
# Parse the response XML
root = ET.fromstring(response.content)
# Find the href element
href_elem = root.find(".//{DAV:}href")
assert href_elem is not None
# Verify the URL is passed through unchanged (sabre/dav should generate it with prefix)
href = href_elem.text
assert href == "/api/v1.0/caldav/calendars/test@example.com/calendar-id/", (
f"Expected URL to be passed through unchanged, got {href}"
)
@responses.activate
def test_proxy_passes_through_namespaced_href_attributes(self):
"""Test that namespaced href attributes (D:href) are passed through unchanged.
Since we removed URL rewriting from the proxy, sabre/dav should generate
URLs with the correct prefix. This test verifies the proxy passes responses through.
"""
user = factories.UserFactory(email="test@example.com")
client = APIClient()
client.force_login(user)
# Mock CalDAV server PROPFIND response with D:href that already has prefix
# (sabre/dav should generate URLs with prefix when X-Forwarded-Prefix is set)
caldav_url = settings.CALDAV_URL
propfind_xml = """<?xml version="1.0"?>
<multistatus xmlns="DAV:" xmlns:D="DAV:">
<response>
<D:href>/api/v1.0/caldav/principals/test@example.com/</D:href>
<propstat>
<prop>
<resourcetype>
<principal/>
</resourcetype>
</prop>
</propstat>
</response>
</multistatus>"""
responses.add(
responses.Response(
method="PROPFIND",
url=f"{caldav_url}/api/v1.0/caldav/",
status=HTTP_207_MULTI_STATUS,
body=propfind_xml,
headers={"Content-Type": "application/xml"},
)
)
response = client.generic("PROPFIND", "/api/v1.0/caldav/")
assert response.status_code == HTTP_207_MULTI_STATUS
# Parse the response XML
root = ET.fromstring(response.content)
# Find the D:href element (namespaced)
href_elem = root.find(".//{DAV:}href")
assert href_elem is not None
# Verify the URL is passed through unchanged (sabre/dav should generate it with prefix)
href = href_elem.text
assert href == "/api/v1.0/caldav/principals/test@example.com/", (
f"Expected URL to be passed through unchanged, got {href}"
)
@responses.activate
def test_proxy_forwards_path_correctly(self):
"""Test that proxy forwards the path correctly to CalDAV server."""
user = factories.UserFactory(email="test@example.com")
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
responses.add(
responses.Response(
method="PROPFIND",
url=f"{caldav_url}/api/v1.0/caldav/principals/test@example.com/",
status=HTTP_207_MULTI_STATUS,
body='<?xml version="1.0"?><multistatus xmlns="DAV:"></multistatus>',
headers={"Content-Type": "application/xml"},
)
)
# Request a specific path
response = client.generic(
"PROPFIND", "/api/v1.0/caldav/principals/test@example.com/"
)
# Verify the request was made to the correct URL
assert len(responses.calls) == 1
request = responses.calls[0].request
assert (
request.url == f"{caldav_url}/api/v1.0/caldav/principals/test@example.com/"
)
@responses.activate
def test_proxy_handles_options_request(self):
"""Test that OPTIONS requests are handled for CORS."""
user = factories.UserFactory(email="test@example.com")
client = APIClient()
client.force_login(user)
response = client.options("/api/v1.0/caldav/")
assert response.status_code == HTTP_200_OK
assert "Access-Control-Allow-Methods" in response
assert "PROPFIND" in response["Access-Control-Allow-Methods"]

35
src/backend/core/tests/test_caldav_service.py
View File

@@ -1,4 +1,4 @@
"""Tests for CalDAV service integration with DAViCal."""
"""Tests for CalDAV service integration."""
from unittest.mock import Mock, patch
@@ -7,17 +7,17 @@ from django.conf import settings
import pytest
from core import factories
from core.services.caldav_service import CalendarService, DAViCalClient
from core.services.caldav_service import CalDAVClient, CalendarService
@pytest.mark.django_db
class TestDAViCalClient:
"""Tests for DAViCalClient authentication and communication."""
class TestCalDAVClient:
"""Tests for CalDAVClient authentication and communication."""
def test_get_client_sends_x_forwarded_user_header(self):
"""Test that DAVClient is configured with X-Forwarded-User header."""
user = factories.UserFactory(email="test@example.com")
client = DAViCalClient()
client = CalDAVClient()
dav_client = client._get_client(user)
@@ -33,16 +33,13 @@ class TestDAViCalClient:
assert dav_client.headers["X-Forwarded-User"] == user.email
@pytest.mark.skipif(
not getattr(settings, "DAVICAL_URL", None),
reason="DAViCal URL not configured",
not settings.CALDAV_URL,
reason="CalDAV server URL not configured",
)
def test_create_calendar_authenticates_with_davical(self):
"""Test that calendar creation authenticates successfully with DAViCal."""
def test_create_calendar_authenticates_with_caldav_server(self):
"""Test that calendar creation authenticates successfully with CalDAV server."""
user = factories.UserFactory(email="test@example.com")
client = DAViCalClient()
# Ensure user exists in DAViCal
client.ensure_user_exists(user)
client = CalDAVClient()
# Try to create a calendar - this should authenticate successfully
calendar_path = client.create_calendar(
@@ -51,11 +48,14 @@ class TestDAViCalClient:
# Verify calendar path was returned
assert calendar_path is not None
assert calendar_path.startswith("/caldav.php/")
assert user.email in calendar_path
# Email may be URL-encoded in the path (e.g., test%40example.com)
assert (
user.email.replace("@", "%40") in calendar_path
or user.email in calendar_path
)
def test_calendar_service_creates_calendar(self):
"""Test that CalendarService can create a calendar through DAViCal."""
"""Test that CalendarService can create a calendar through CalDAV server."""
user = factories.UserFactory(email="test@example.com")
service = CalendarService()
@@ -67,5 +67,4 @@ class TestDAViCalClient:
assert calendar.owner == user
assert calendar.name == "My Calendar"
assert calendar.color == "#ff0000"
assert calendar.davical_path is not None
assert calendar.davical_path.startswith("/caldav.php/")
assert calendar.caldav_path is not None