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

✨(freebusy) add availability management (#35)

Browse Source 
Adds organization-level default calendar sharing controls, "Find a Time" scheduling UI with a Free/Busy timeline showing attendee availability and conflicts, Working hours editor in Settings to manage and save availability, Autocomplete attendee search with debounced, partial name/email matching and timezone display.

Fixes #26. Fixes #25. Fixes #24.
This commit is contained in:
Sylvain Zimmer
2026-03-10 01:30:42 +01:00
committed by GitHub
parent 9c18f96090
commit 7cb8d5e7b6
49 changed files with 3714 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/backend/calendars/settings.py
View File

@@ -83,6 +83,15 @@ class Base(Configuration):
CALDAV_INTERNAL_API_KEY = SecretFileValue(
None, environ_name="CALDAV_INTERNAL_API_KEY", environ_prefix=None
)
# Default calendar sharing level for new organizations.
# Controls what colleagues in the same org can see by default.
# Values: "none", "freebusy", "read", "write"
ORG_DEFAULT_SHARING_LEVEL = values.Value(
"freebusy",
environ_name="ORG_DEFAULT_SHARING_LEVEL",
environ_prefix=None,
)
# Salt for django-fernet-encrypted-fields (Channel tokens, etc.)
# Used with SECRET_KEY to derive Fernet encryption keys via PBKDF2
SALT_KEY = values.Value(
@@ -931,8 +940,8 @@ class Development(Base):
EMAIL_PORT = 1025
EMAIL_USE_TLS = False
EMAIL_USE_SSL = False
DEFAULT_FROM_EMAIL = "calendars@calendars.world"
CALENDAR_INVITATION_FROM_EMAIL = "calendars@calendars.world"
DEFAULT_FROM_EMAIL = "noreply@example.local"
CALENDAR_INVITATION_FROM_EMAIL = "noreply@example.local"
APP_NAME = "Calendars (dev)"
APP_URL = "http://localhost:8931"

13
src/backend/core/api/serializers.py
View File

@@ -4,6 +4,7 @@ from django.conf import settings
from django.utils.text import slugify
from rest_framework import serializers
from timezone_field.rest_framework import TimeZoneSerializerField
from core import models
from core.entitlements import EntitlementsUnavailableError, get_user_entitlements
@@ -13,10 +14,16 @@ from core.models import uuid_to_urlsafe
class OrganizationSerializer(serializers.ModelSerializer):
"""Serialize organizations."""
sharing_level = serializers.SerializerMethodField(read_only=True)
class Meta:
model = models.Organization
fields = ["id", "name"]
read_only_fields = ["id", "name"]
fields = ["id", "name", "sharing_level"]
read_only_fields = ["id", "name", "sharing_level"]
def get_sharing_level(self, org) -> str:
"""Return the effective sharing level (org override or server default)."""
return org.effective_sharing_level
class UserLiteSerializer(serializers.ModelSerializer):
@@ -32,6 +39,7 @@ class UserSerializer(serializers.ModelSerializer):
"""Serialize users."""
email = serializers.SerializerMethodField(read_only=True)
timezone = TimeZoneSerializerField(use_pytz=False)
class Meta:
model = models.User
@@ -40,6 +48,7 @@ class UserSerializer(serializers.ModelSerializer):
"email",
"full_name",
"language",
"timezone",
]
read_only_fields = ["id", "email", "full_name"]

68
src/backend/core/api/viewsets.py
View File

@@ -5,6 +5,7 @@ import logging
from django.conf import settings
from django.core.cache import cache
from django.db.models import Q
from django.utils.text import slugify
from rest_framework import mixins, pagination, response, status, views, viewsets
@@ -98,9 +99,9 @@ class UserViewSet(
def get_queryset(self):
"""
Limit listed users by querying the email field.
Limit listed users by querying email or full_name.
Scoped to the requesting user's organization.
If query contains "@", search exactly. Otherwise return empty.
Minimum 3 characters required.
"""
queryset = self.queryset
@@ -112,15 +113,13 @@ class UserViewSet(
return queryset.none()
queryset = queryset.filter(organization_id=self.request.user.organization_id)
if not (query := self.request.query_params.get("q", "")) or len(query) < 5:
if not (query := self.request.query_params.get("q", "")) or len(query) < 3:
return queryset.none()
# For emails, match exactly
if "@" in query:
return queryset.filter(email__iexact=query).order_by("email")
# For non-email queries, return empty (no fuzzy search)
return queryset.none()
# Search by email (partial, case-insensitive) or full name
return queryset.filter(
Q(email__icontains=query) | Q(full_name__icontains=query)
).order_by("full_name", "email")[:50]
@action(
detail=False,
@@ -210,6 +209,57 @@ class ConfigView(views.APIView):
return theme_customization
class OrganizationSettingsViewSet(viewsets.ViewSet):
"""ViewSet for organization settings (sharing level, etc.).
Only org admins can update settings; all org members can read them.
"""
permission_classes = [IsAuthenticated]
def retrieve(self, request, pk=None): # pylint: disable=unused-argument
"""GET /api/v1.0/organization-settings/current/"""
org = request.user.organization
if not org:
return response.Response(
{"detail": "User has no organization."},
status=status.HTTP_404_NOT_FOUND,
)
return response.Response(serializers.OrganizationSerializer(org).data)
def partial_update(self, request, pk=None): # pylint: disable=unused-argument
"""PATCH /api/v1.0/organization-settings/current/"""
if not request.user.organization:
return response.Response(
{"detail": "User has no organization."},
status=status.HTTP_404_NOT_FOUND,
)
# Check admin permission
perm = permissions.IsOrgAdmin()
if not perm.has_permission(request, self):
return response.Response(
{"detail": "Only org admins can update settings."},
status=status.HTTP_403_FORBIDDEN,
)
org = request.user.organization
sharing_level = request.data.get("default_sharing_level")
if sharing_level is not None:
valid = {c[0] for c in models.SharingLevel.choices}
if sharing_level not in valid:
return response.Response(
{
"detail": f"Invalid sharing level. Must be one of: {', '.join(valid)}"
},
status=status.HTTP_400_BAD_REQUEST,
)
org.default_sharing_level = sharing_level
org.save(update_fields=["default_sharing_level", "updated_at"])
return response.Response(serializers.OrganizationSerializer(org).data)
class CalendarViewSet(viewsets.GenericViewSet):
"""ViewSet for calendar operations.

18
src/backend/core/migrations/0002_organization_default_sharing_level.py Normal file
View File

@@ -0,0 +1,18 @@
# Generated by Django 5.2.9 on 2026-03-09 08:25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='organization',
name='default_sharing_level',
field=models.CharField(blank=True, choices=[('none', 'No sharing'), ('freebusy', 'Free/Busy only'), ('read', 'Read access'), ('write', 'Read/Write access')], help_text='Default calendar sharing level for org members. Null means use the server-wide default (ORG_DEFAULT_SHARING_LEVEL).', max_length=10, null=True),
),
]

26
src/backend/core/models.py
View File

@@ -66,6 +66,15 @@ class BaseModel(models.Model):
super().save(*args, **kwargs)
class SharingLevel(models.TextChoices):
"""Calendar sharing visibility levels within an organization."""
NONE = "none", "No sharing"
FREEBUSY = "freebusy", "Free/Busy only"
READ = "read", "Read access"
WRITE = "write", "Read/Write access"
class Organization(BaseModel):
"""Organization model, populated from OIDC claims and entitlements.
@@ -81,6 +90,16 @@ class Organization(BaseModel):
db_index=True,
help_text="Organization identifier from OIDC claim or email domain.",
)
default_sharing_level = models.CharField(
max_length=10,
choices=SharingLevel.choices,
null=True,
blank=True,
help_text=(
"Default calendar sharing level for org members. "
"Null means use the server-wide default (ORG_DEFAULT_SHARING_LEVEL)."
),
)
class Meta:
db_table = "calendars_organization"
@@ -90,6 +109,13 @@ class Organization(BaseModel):
def __str__(self):
return self.name or self.external_id
@property
def effective_sharing_level(self):
"""Return the effective sharing level, falling back to server default."""
if self.default_sharing_level:
return self.default_sharing_level
return settings.ORG_DEFAULT_SHARING_LEVEL
def delete(self, *args, **kwargs):
"""Delete org after cleaning up members' CalDAV data.

6
src/backend/core/services/caldav_service.py
View File

@@ -58,11 +58,15 @@ class CalDAVHTTPClient:
"""
if not user.email:
raise ValueError("User has no email address")
return {
headers = {
"X-Api-Key": cls.get_api_key(),
"X-Forwarded-User": user.email,
"X-CalDAV-Organization": str(user.organization_id),
}
org = getattr(user, "organization", None)
if org and hasattr(org, "effective_sharing_level"):
headers["X-CalDAV-Sharing-Level"] = org.effective_sharing_level
return headers
def build_url(self, path: str, query: str = "") -> str:
"""Build a full CalDAV URL from a resource path.

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

@@ -73,61 +73,38 @@ def test_api_users_list_query_inactive():
def test_api_users_list_query_short_queries():
"""
Queries shorter than 5 characters should return an empty result set.
"""
"""Queries shorter than 3 characters should return an empty result set."""
user = factories.UserFactory()
org = user.organization
client = APIClient()
client.force_login(user)
factories.UserFactory(email="john.doe@example.com")
factories.UserFactory(email="john.lennon@example.com")
factories.UserFactory(email="john.doe@example.com", organization=org)
response = client.get("/api/v1.0/users/?q=jo")
assert response.status_code == 200
assert response.json()["results"] == []
response = client.get("/api/v1.0/users/?q=john")
assert response.status_code == 200
assert response.json()["results"] == []
# Non-email queries (without @) return empty
response = client.get("/api/v1.0/users/?q=john.")
response = client.get("/api/v1.0/users/?q=j")
assert response.status_code == 200
assert response.json()["results"] == []
def test_api_users_list_limit(settings):
"""
Authenticated users should be able to list users and the number of results
should be limited to 10.
"""
def test_api_users_list_limit(settings): # pylint: disable=unused-argument
"""Results should be bounded even with many matching users."""
user = factories.UserFactory()
org = user.organization
client = APIClient()
client.force_login(user)
# Use a base name with a length equal 5 to test that the limit is applied
base_name = "alice"
for i in range(15):
factories.UserFactory(email=f"{base_name}.{i}@example.com", organization=org)
for i in range(55):
factories.UserFactory(email=f"alice.{i}@example.com", organization=org)
# Non-email queries (without @) return empty
response = client.get(
"/api/v1.0/users/?q=alice",
)
# Partial match returns results (capped at 50)
response = client.get("/api/v1.0/users/?q=alice")
assert response.status_code == 200
assert response.json()["results"] == []
# Email queries require exact match
settings.API_USERS_LIST_LIMIT = 100
response = client.get(
"/api/v1.0/users/?q=alice.0@example.com",
)
assert response.status_code == 200
assert len(response.json()["results"]) == 1
assert len(response.json()["results"]) == 50
def test_api_users_list_throttling_authenticated(settings):
@@ -154,8 +131,7 @@ def test_api_users_list_throttling_authenticated(settings):
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).
Authenticated users should be able to search users by partial email.
"""
settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"]["user_list_burst"] = "9999/minute"
@@ -167,42 +143,35 @@ def test_api_users_list_query_email(settings):
client.force_login(user)
dave = factories.UserFactory(email="david.bowman@work.com", organization=org)
factories.UserFactory(email="nicole.bowman@work.com", organization=org)
nicole = factories.UserFactory(email="nicole.bowman@work.com", organization=org)
# Exact match works
response = client.get(
"/api/v1.0/users/?q=david.bowman@work.com",
)
response = client.get("/api/v1.0/users/?q=david.bowman@work.com")
assert response.status_code == 200
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids == [str(dave.id)]
user_ids = [u["id"] for u in response.json()["results"]]
assert str(dave.id) in user_ids
# Partial email match works
response = client.get("/api/v1.0/users/?q=bowman@work")
assert response.status_code == 200
user_ids = [u["id"] for u in response.json()["results"]]
assert str(dave.id) in user_ids
assert str(nicole.id) in user_ids
# Case-insensitive match works
response = client.get(
"/api/v1.0/users/?q=David.Bowman@Work.COM",
)
response = client.get("/api/v1.0/users/?q=David.Bowman@Work.COM")
assert response.status_code == 200
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids == [str(dave.id)]
user_ids = [u["id"] for u in response.json()["results"]]
assert str(dave.id) in user_ids
# Typos don't match (exact match only)
response = client.get(
"/api/v1.0/users/?q=davig.bovman@worm.com",
)
# Typos don't match
response = client.get("/api/v1.0/users/?q=davig.bovman@worm.com")
assert response.status_code == 200
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids == []
response = client.get(
"/api/v1.0/users/?q=davig.bovman@worm.cop",
)
assert response.status_code == 200
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids == []
assert response.json()["results"] == []
def test_api_users_list_query_email_matching():
"""Email queries return exact matches only (case-insensitive)."""
def test_api_users_list_query_email_partial_matching():
"""Partial email queries return matching users."""
user = factories.UserFactory()
org = user.organization
@@ -212,29 +181,111 @@ def test_api_users_list_query_email_matching():
user1 = factories.UserFactory(
email="alice.johnson@example.gouv.fr", organization=org
)
factories.UserFactory(email="alice.johnnson@example.gouv.fr", organization=org)
user2 = factories.UserFactory(
email="alice.johnnson@example.gouv.fr", organization=org
)
factories.UserFactory(email="alice.kohlson@example.gouv.fr", organization=org)
user4 = factories.UserFactory(
email="alicia.johnnson@example.gouv.fr", organization=org
)
factories.UserFactory(email="alicia.johnnson@example.gov.uk", organization=org)
# Different org user should not appear
other_org_user = factories.UserFactory(email="alice.johnnson@example.gov.uk")
factories.UserFactory(email="alice.thomson@example.gouv.fr", organization=org)
# Exact match returns only that user
response = client.get(
"/api/v1.0/users/?q=alice.johnson@example.gouv.fr",
)
# Partial match on "alice.john" returns alice.johnson and alice.johnnson
response = client.get("/api/v1.0/users/?q=alice.john")
assert response.status_code == 200
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids == [str(user1.id)]
user_ids = [u["id"] for u in response.json()["results"]]
assert str(user1.id) in user_ids
assert str(user2.id) in user_ids
assert str(other_org_user.id) not in user_ids
# Different email returns different user
response = client.get("/api/v1.0/users/?q=alicia.johnnson@example.gouv.fr")
# Partial match on "alicia" returns alicia.johnnson (same org only)
response = client.get("/api/v1.0/users/?q=alicia")
assert response.status_code == 200
user_ids = [user["id"] for user in response.json()["results"]]
user_ids = [u["id"] for u in response.json()["results"]]
assert user_ids == [str(user4.id)]
def test_api_users_list_query_by_name():
"""Users should be searchable by full name (partial, case-insensitive)."""
user = factories.UserFactory()
org = user.organization
client = APIClient()
client.force_login(user)
alice = factories.UserFactory(
email="alice@example.com", full_name="Alice Johnson", organization=org
)
bob = factories.UserFactory(
email="bob@example.com", full_name="Bob Smith", organization=org
)
factories.UserFactory(
email="charlie@example.com", full_name="Charlie Johnson", organization=org
)
# Search by first name
response = client.get("/api/v1.0/users/?q=Alice")
assert response.status_code == 200
user_ids = [u["id"] for u in response.json()["results"]]
assert str(alice.id) in user_ids
assert str(bob.id) not in user_ids
# Search by last name matches multiple users
response = client.get("/api/v1.0/users/?q=Johnson")
assert response.status_code == 200
assert len(response.json()["results"]) == 2
# Case-insensitive
response = client.get("/api/v1.0/users/?q=bob")
assert response.status_code == 200
user_ids = [u["id"] for u in response.json()["results"]]
assert str(bob.id) in user_ids
def test_api_users_list_cross_org_isolation():
"""Users from different organizations should not see each other."""
org1 = factories.OrganizationFactory(name="Org One")
org2 = factories.OrganizationFactory(name="Org Two")
user1 = factories.UserFactory(
email="user1@org1.com", full_name="Shared Name", organization=org1
)
factories.UserFactory(
email="user2@org2.com", full_name="Shared Name", organization=org2
)
client = APIClient()
client.force_login(user1)
# Search by shared name - should only return same-org user
response = client.get("/api/v1.0/users/?q=Shared")
assert response.status_code == 200
user_ids = [u["id"] for u in response.json()["results"]]
assert str(user1.id) in user_ids
assert len(user_ids) == 1
# Search by cross-org email - should return nothing
response = client.get("/api/v1.0/users/?q=user2@org2.com")
assert response.status_code == 200
assert response.json()["results"] == []
def test_api_users_list_includes_self():
"""Search should include the requesting user if they match."""
user = factories.UserFactory(email="alice@example.com", full_name="Alice Test")
client = APIClient()
client.force_login(user)
# User should find themselves
response = client.get("/api/v1.0/users/?q=alice")
assert response.status_code == 200
user_ids = [u["id"] for u in response.json()["results"]]
assert str(user.id) in user_ids
def test_api_users_retrieve_me_anonymous():
"""Anonymous users should not be allowed to list users."""
factories.UserFactory.create_batch(2)
@@ -271,11 +322,13 @@ def test_api_users_retrieve_me_authenticated():
"email": user.email,
"full_name": user.full_name,
"language": user.language,
"timezone": str(user.timezone),
"can_access": True,
"can_admin": True,
"organization": {
"id": str(user.organization.id),
"name": user.organization.name,
"sharing_level": "freebusy",
},
}
@@ -440,8 +493,8 @@ def test_api_users_update_anonymous():
def test_api_users_update_authenticated_self():
"""
Authenticated users should be able to update their own user but only "language"
and "timezone" fields.
Authenticated users should be able to update their own user but only "language",
"timezone" fields.
"""
user = factories.UserFactory()
@@ -528,8 +581,8 @@ def test_api_users_patch_anonymous():
def test_api_users_patch_authenticated_self():
"""
Authenticated users should be able to patch their own user but only "language"
and "timezone" fields.
Authenticated users should be able to patch their own user but only "language",
"timezone" fields.
"""
user = factories.UserFactory()

226
src/backend/core/tests/test_caldav_proxy.py
View File

@@ -335,6 +335,232 @@ class TestCalDAVProxy:
assert response.status_code == HTTP_400_BAD_REQUEST
@pytest.mark.django_db
class TestCalDAVFreeBusy:
"""Tests for free/busy queries via CalDAV outbox POST."""
FREEBUSY_REQUEST = (
"BEGIN:VCALENDAR\r\n"
"VERSION:2.0\r\n"
"PRODID:-//Test//EN\r\n"
"METHOD:REQUEST\r\n"
"BEGIN:VFREEBUSY\r\n"
"DTSTART:20260309T000000Z\r\n"
"DTEND:20260310T000000Z\r\n"
"ORGANIZER:mailto:{organizer}\r\n"
"ATTENDEE:mailto:{attendee}\r\n"
"END:VFREEBUSY\r\n"
"END:VCALENDAR"
)
FREEBUSY_RESPONSE = (
'<?xml version="1.0" encoding="utf-8"?>\n'
'<cal:schedule-response xmlns:d="DAV:" '
'xmlns:cal="urn:ietf:params:xml:ns:caldav">\n'
" <cal:response>\n"
" <cal:recipient><d:href>mailto:{attendee}</d:href></cal:recipient>\n"
" <cal:request-status>2.0;Success</cal:request-status>\n"
" <cal:calendar-data>"
"BEGIN:VCALENDAR\r\n"
"VERSION:2.0\r\n"
"PRODID:-//SabreDAV//EN\r\n"
"BEGIN:VFREEBUSY\r\n"
"DTSTART:20260309T000000Z\r\n"
"DTEND:20260310T000000Z\r\n"
"FREEBUSY:20260309T100000Z/20260309T110000Z\r\n"
"END:VFREEBUSY\r\n"
"END:VCALENDAR"
"</cal:calendar-data>\n"
" </cal:response>\n"
"</cal:schedule-response>"
)
@responses.activate
def test_freebusy_post_forwarded_with_correct_content_type(self):
"""POST to outbox should forward text/calendar content-type to CalDAV."""
user = factories.UserFactory(email="alice@example.com")
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
outbox_path = f"calendars/users/{user.email}/outbox/"
responses.add(
responses.Response(
method="POST",
url=f"{caldav_url}/caldav/{outbox_path}",
status=HTTP_200_OK,
body=self.FREEBUSY_RESPONSE.format(attendee="bob@example.com"),
headers={"Content-Type": "application/xml"},
)
)
body = self.FREEBUSY_REQUEST.format(
organizer=user.email, attendee="bob@example.com"
)
response = client.generic(
"POST",
f"/caldav/{outbox_path}",
data=body,
content_type="text/calendar; charset=utf-8",
)
assert response.status_code == HTTP_200_OK
assert len(responses.calls) == 1
# Verify content-type is forwarded (not overwritten to application/xml)
forwarded = responses.calls[0].request
assert "text/calendar" in forwarded.headers["Content-Type"]
@responses.activate
def test_freebusy_post_forwards_body(self):
"""POST to outbox should forward the iCalendar body unchanged."""
user = factories.UserFactory(email="alice@example.com")
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
outbox_path = f"calendars/users/{user.email}/outbox/"
responses.add(
responses.Response(
method="POST",
url=f"{caldav_url}/caldav/{outbox_path}",
status=HTTP_200_OK,
body=self.FREEBUSY_RESPONSE.format(attendee="bob@example.com"),
headers={"Content-Type": "application/xml"},
)
)
body = self.FREEBUSY_REQUEST.format(
organizer=user.email, attendee="bob@example.com"
)
client.generic(
"POST",
f"/caldav/{outbox_path}",
data=body,
content_type="text/calendar; charset=utf-8",
)
# Verify the body was forwarded
forwarded = responses.calls[0].request
assert b"BEGIN:VCALENDAR" in forwarded.body
assert b"VFREEBUSY" in forwarded.body
assert b"bob@example.com" in forwarded.body
@responses.activate
def test_freebusy_post_forwards_auth_headers(self):
"""POST to outbox should include X-Forwarded-User and X-Api-Key."""
user = factories.UserFactory(email="alice@example.com")
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
outbox_path = f"calendars/users/{user.email}/outbox/"
responses.add(
responses.Response(
method="POST",
url=f"{caldav_url}/caldav/{outbox_path}",
status=HTTP_200_OK,
body=self.FREEBUSY_RESPONSE.format(attendee="bob@example.com"),
headers={"Content-Type": "application/xml"},
)
)
body = self.FREEBUSY_REQUEST.format(
organizer=user.email, attendee="bob@example.com"
)
client.generic(
"POST",
f"/caldav/{outbox_path}",
data=body,
content_type="text/calendar; charset=utf-8",
)
forwarded = responses.calls[0].request
assert forwarded.headers["X-Forwarded-User"] == user.email
assert forwarded.headers["X-Api-Key"] == settings.CALDAV_OUTBOUND_API_KEY
@responses.activate
def test_freebusy_post_returns_schedule_response(self):
"""POST to outbox should return the CalDAV schedule-response XML."""
user = factories.UserFactory(email="alice@example.com")
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
outbox_path = f"calendars/users/{user.email}/outbox/"
response_body = self.FREEBUSY_RESPONSE.format(attendee="bob@example.com")
responses.add(
responses.Response(
method="POST",
url=f"{caldav_url}/caldav/{outbox_path}",
status=HTTP_200_OK,
body=response_body,
headers={"Content-Type": "application/xml"},
)
)
body = self.FREEBUSY_REQUEST.format(
organizer=user.email, attendee="bob@example.com"
)
response = client.generic(
"POST",
f"/caldav/{outbox_path}",
data=body,
content_type="text/calendar; charset=utf-8",
)
assert response.status_code == HTTP_200_OK
# Verify the schedule-response is returned to the client
root = ET.fromstring(response.content)
ns = {"cal": "urn:ietf:params:xml:ns:caldav", "d": "DAV:"}
status = root.find(".//cal:request-status", ns)
assert status is not None
assert "2.0" in status.text
def test_freebusy_post_requires_authentication(self):
"""POST to outbox should require authentication."""
client = APIClient()
response = client.generic(
"POST",
"/caldav/calendars/users/alice@example.com/outbox/",
data="BEGIN:VCALENDAR\r\nEND:VCALENDAR",
content_type="text/calendar",
)
assert response.status_code == HTTP_401_UNAUTHORIZED
@responses.activate
def test_freebusy_post_includes_organization_header(self):
"""POST to outbox should include X-CalDAV-Organization header."""
user = factories.UserFactory(email="alice@example.com")
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
outbox_path = f"calendars/users/{user.email}/outbox/"
responses.add(
responses.Response(
method="POST",
url=f"{caldav_url}/caldav/{outbox_path}",
status=HTTP_200_OK,
body=self.FREEBUSY_RESPONSE.format(attendee="bob@example.com"),
headers={"Content-Type": "application/xml"},
)
)
body = self.FREEBUSY_REQUEST.format(
organizer=user.email, attendee="bob@example.com"
)
client.generic(
"POST",
f"/caldav/{outbox_path}",
data=body,
content_type="text/calendar; charset=utf-8",
)
forwarded = responses.calls[0].request
assert forwarded.headers["X-CalDAV-Organization"] == str(user.organization_id)
class TestValidateCaldavProxyPath:
"""Tests for validate_caldav_proxy_path utility."""

97
src/backend/core/tests/test_organizations.py
View File

@@ -194,3 +194,100 @@ def test_user_list_same_org_visible():
assert len(data) == 1
assert data[0]["email"] == "carol@example.com"
get_entitlements_backend.cache_clear()
# -- Sharing level --
@pytest.mark.django_db
def test_effective_sharing_level_defaults_to_server_setting():
"""Organization without override returns server-wide default."""
org = factories.OrganizationFactory(default_sharing_level=None)
assert org.effective_sharing_level == "freebusy"
@pytest.mark.django_db
@override_settings(ORG_DEFAULT_SHARING_LEVEL="none")
def test_effective_sharing_level_follows_server_override():
"""Organization without override returns overridden server default."""
org = factories.OrganizationFactory(default_sharing_level=None)
assert org.effective_sharing_level == "none"
@pytest.mark.django_db
def test_effective_sharing_level_org_override():
"""Organization with explicit level ignores server default."""
org = factories.OrganizationFactory(default_sharing_level="read")
assert org.effective_sharing_level == "read"
# -- Organization settings API --
@pytest.mark.django_db
@override_settings(
ENTITLEMENTS_BACKEND="core.entitlements.backends.local.LocalEntitlementsBackend",
ENTITLEMENTS_BACKEND_PARAMETERS={},
)
def test_org_settings_retrieve():
"""GET /organization-settings/current/ returns org sharing level."""
get_entitlements_backend.cache_clear()
org = factories.OrganizationFactory(external_id="test-org")
user = factories.UserFactory(organization=org)
client = APIClient()
client.force_authenticate(user=user)
response = client.get("/api/v1.0/organization-settings/current/")
assert response.status_code == HTTP_200_OK
assert response.json()["sharing_level"] == "freebusy"
get_entitlements_backend.cache_clear()
@pytest.mark.django_db
@override_settings(
ENTITLEMENTS_BACKEND="core.entitlements.backends.local.LocalEntitlementsBackend",
ENTITLEMENTS_BACKEND_PARAMETERS={},
)
def test_org_settings_update_sharing_level():
"""PATCH /organization-settings/current/ updates sharing level."""
get_entitlements_backend.cache_clear()
org = factories.OrganizationFactory(external_id="test-org")
user = factories.UserFactory(organization=org)
client = APIClient()
client.force_authenticate(user=user)
response = client.patch(
"/api/v1.0/organization-settings/current/",
{"default_sharing_level": "none"},
format="json",
)
assert response.status_code == HTTP_200_OK
org.refresh_from_db()
assert org.default_sharing_level == "none"
assert response.json()["sharing_level"] == "none"
get_entitlements_backend.cache_clear()
@pytest.mark.django_db
@override_settings(
ENTITLEMENTS_BACKEND="core.entitlements.backends.local.LocalEntitlementsBackend",
ENTITLEMENTS_BACKEND_PARAMETERS={},
)
def test_org_settings_update_rejects_invalid_level():
"""PATCH with invalid sharing level returns 400."""
get_entitlements_backend.cache_clear()
org = factories.OrganizationFactory(external_id="test-org")
user = factories.UserFactory(organization=org)
client = APIClient()
client.force_authenticate(user=user)
response = client.patch(
"/api/v1.0/organization-settings/current/",
{"default_sharing_level": "superadmin"},
format="json",
)
assert response.status_code == 400
get_entitlements_backend.cache_clear()

28
src/backend/core/tests/test_permissions_security.py
View File

@@ -272,6 +272,34 @@ class TestCalDAVProxyOrgHeader:
assert request.headers["X-CalDAV-Organization"] == str(org.id)
assert request.headers["X-CalDAV-Organization"] != "spoofed-org-id"
@responses.activate
def test_proxy_sends_sharing_level_header(self):
"""CalDAV proxy sends X-CalDAV-Sharing-Level from org's effective level."""
org = factories.OrganizationFactory(
external_id="org-fb", default_sharing_level="none"
)
user = factories.UserFactory(email="alice@example.com", organization=org)
client = APIClient()
client.force_login(user)
caldav_url = settings.CALDAV_URL
responses.add(
responses.Response(
method="PROPFIND",
url=f"{caldav_url}/caldav/principals/resources/",
status=HTTP_207_MULTI_STATUS,
body='<?xml version="1.0"?><multistatus xmlns="DAV:"></multistatus>',
headers={"Content-Type": "application/xml"},
)
)
client.generic("PROPFIND", "/caldav/principals/resources/")
assert len(responses.calls) == 1
request = responses.calls[0].request
assert request.headers["X-CalDAV-Sharing-Level"] == "none"
# ---------------------------------------------------------------------------
# IsEntitledToAccess permission — fail-closed

5
src/backend/core/urls.py
View File

@@ -20,6 +20,11 @@ router.register("users", viewsets.UserViewSet, basename="users")
router.register("calendars", viewsets.CalendarViewSet, basename="calendars")
router.register("resources", viewsets.ResourceViewSet, basename="resources")
router.register("channels", ChannelViewSet, basename="channels")
router.register(
"organization-settings",
viewsets.OrganizationSettingsViewSet,
basename="organization-settings",
)
urlpatterns = [
path(