(back) add subscription and iCal API

Add API endpoints for calendar subscription token management
and iCal export. Includes serializers, viewsets and URL
configuration for subscription URLs and .ics file generation.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Nathan Panchout
2026-01-25 20:33:11 +01:00
parent 7e90c960dc
commit f329de9354
6 changed files with 486 additions and 16 deletions

View File

@@ -81,6 +81,44 @@ class Base(Configuration):
CALDAV_OUTBOUND_API_KEY = values.Value(
None, environ_name="CALDAV_OUTBOUND_API_KEY", environ_prefix=None
)
# Base URL for CalDAV scheduling callbacks (must be accessible from CalDAV container)
# In Docker environments, use the internal Docker network URL (e.g., http://backend:8000)
CALDAV_CALLBACK_BASE_URL = values.Value(
None, environ_name="CALDAV_CALLBACK_BASE_URL", environ_prefix=None
)
# Email configuration
# Default settings - override in environment-specific classes
EMAIL_BACKEND = values.Value(
"django.core.mail.backends.smtp.EmailBackend",
environ_name="EMAIL_BACKEND",
environ_prefix=None,
)
EMAIL_HOST = values.Value(
"localhost", environ_name="EMAIL_HOST", environ_prefix=None
)
EMAIL_PORT = values.IntegerValue(25, environ_name="EMAIL_PORT", environ_prefix=None)
EMAIL_HOST_USER = values.Value(
"", environ_name="EMAIL_HOST_USER", environ_prefix=None
)
EMAIL_HOST_PASSWORD = SecretFileValue(
"", environ_name="EMAIL_HOST_PASSWORD", environ_prefix=None
)
EMAIL_USE_TLS = values.BooleanValue(
False, environ_name="EMAIL_USE_TLS", environ_prefix=None
)
EMAIL_USE_SSL = values.BooleanValue(
False, environ_name="EMAIL_USE_SSL", environ_prefix=None
)
DEFAULT_FROM_EMAIL = values.Value(
"noreply@example.com", environ_name="DEFAULT_FROM_EMAIL", environ_prefix=None
)
# Calendar-specific email settings
CALENDAR_INVITATION_FROM_EMAIL = values.Value(
None, environ_name="CALENDAR_INVITATION_FROM_EMAIL", environ_prefix=None
)
APP_NAME = values.Value("Calendrier", environ_name="APP_NAME", environ_prefix=None)
APP_URL = values.Value("", environ_name="APP_URL", environ_prefix=None)
# Security
ALLOWED_HOSTS = values.ListValue([])
@@ -354,7 +392,7 @@ class Base(Configuration):
CORS_ALLOW_ALL_ORIGINS = values.BooleanValue(False)
CORS_ALLOWED_ORIGINS = values.ListValue([])
CORS_ALLOWED_ORIGIN_REGEXES = values.ListValue([])
# Allow CalDAV methods (PROPFIND, REPORT, etc.)
# Allow CalDAV methods (PROPFIND, PROPPATCH, REPORT, etc.)
CORS_ALLOW_METHODS = [
"DELETE",
"GET",
@@ -363,6 +401,7 @@ class Base(Configuration):
"POST",
"PUT",
"PROPFIND",
"PROPPATCH",
"REPORT",
"MKCOL",
"MKCALENDAR",
@@ -801,6 +840,7 @@ class Development(Base):
CORS_ALLOW_ALL_ORIGINS = True
CSRF_TRUSTED_ORIGINS = [
"http://localhost:8920",
"http://localhost:3000",
]
DEBUG = True
LOAD_E2E_URLS = True
@@ -809,6 +849,16 @@ class Development(Base):
USE_SWAGGER = True
# Email settings for development (mailcatcher)
EMAIL_HOST = "mailcatcher"
EMAIL_PORT = 1025
EMAIL_USE_TLS = False
EMAIL_USE_SSL = False
DEFAULT_FROM_EMAIL = "calendars@calendars.world"
CALENDAR_INVITATION_FROM_EMAIL = "calendars@calendars.world"
APP_NAME = "Calendrier (Dev)"
APP_URL = "http://localhost:8920"
DEBUG_TOOLBAR_CONFIG = {
"SHOW_TOOLBAR_CALLBACK": lambda request: True,
}

View File

@@ -153,3 +153,63 @@ class CalendarShareSerializer(serializers.ModelSerializer):
model = models.CalendarShare
fields = ["id", "shared_with_email", "permission", "is_visible", "created_at"]
read_only_fields = ["id", "created_at"]
class CalendarSubscriptionTokenSerializer(serializers.ModelSerializer):
"""Serializer for CalendarSubscriptionToken model."""
url = serializers.SerializerMethodField()
class Meta:
model = models.CalendarSubscriptionToken
fields = [
"token",
"url",
"caldav_path",
"calendar_name",
"is_active",
"last_accessed_at",
"created_at",
]
read_only_fields = [
"token",
"url",
"caldav_path",
"calendar_name",
"is_active",
"last_accessed_at",
"created_at",
]
def get_url(self, obj) -> str:
"""Build the full subscription URL, enforcing HTTPS in production."""
request = self.context.get("request")
if request:
url = request.build_absolute_uri(f"/ical/{obj.token}.ics")
else:
# Fallback to APP_URL if no request context
app_url = getattr(settings, "APP_URL", "")
url = f"{app_url.rstrip('/')}/ical/{obj.token}.ics"
# Force HTTPS in production to protect the token in transit
if not settings.DEBUG and url.startswith("http://"):
url = url.replace("http://", "https://", 1)
return url
class CalendarSubscriptionTokenCreateSerializer(serializers.Serializer):
"""Serializer for creating a CalendarSubscriptionToken."""
caldav_path = serializers.CharField(max_length=512)
calendar_name = serializers.CharField(max_length=255, required=False, default="")
def validate_caldav_path(self, value):
"""Validate and normalize the caldav_path."""
# Normalize path to always have trailing slash
if not value.endswith("/"):
value = value + "/"
# Normalize path to always start with /
if not value.startswith("/"):
value = "/" + value
return value

View File

@@ -395,3 +395,146 @@ class CalendarViewSet(
serializers.CalendarShareSerializer(share).data,
status=status.HTTP_201_CREATED if created else status.HTTP_200_OK,
)
class SubscriptionTokenViewSet(viewsets.GenericViewSet):
"""
ViewSet for managing subscription tokens independently of Django Calendar model.
This viewset operates directly with CalDAV paths, without requiring a Django
Calendar record. The backend verifies that the user has access to the calendar
by checking that their email is in the CalDAV path.
Endpoints:
- POST /api/v1.0/subscription-tokens/ - Create or get existing token
- GET /api/v1.0/subscription-tokens/by-path/ - Get token by CalDAV path
- DELETE /api/v1.0/subscription-tokens/by-path/ - Delete token by CalDAV path
"""
permission_classes = [IsAuthenticated]
serializer_class = serializers.CalendarSubscriptionTokenSerializer
# Regex for CalDAV path validation
# Pattern: /calendars/<email>/<calendar-id>/
# Calendar ID: alphanumeric with hyphens only (prevents path traversal like ../)
# This blocks injection attacks while allowing UUIDs and test identifiers
CALDAV_PATH_PATTERN = re.compile(
r"^/calendars/[^/]+/[a-zA-Z0-9-]+/$",
)
def _verify_caldav_access(self, user, caldav_path):
"""
Verify that the user has access to the CalDAV calendar.
We verify by checking:
1. The path matches the expected pattern (prevents path injection)
2. The user's email matches the email in the path
CalDAV paths follow the pattern: /calendars/<user_email>/<calendar_id>/
"""
# Format validation to prevent path injection attacks (e.g., ../, query params)
if not self.CALDAV_PATH_PATTERN.match(caldav_path):
logger.warning(
"Invalid CalDAV path format rejected: %s",
caldav_path[:100], # Truncate for logging
)
return False
# Extract and verify email from path
# Path format: /calendars/user@example.com/calendar-id/
parts = caldav_path.strip("/").split("/")
if len(parts) >= 2 and parts[0] == "calendars":
path_email = unquote(parts[1])
return path_email.lower() == user.email.lower()
return False
def _normalize_caldav_path(self, caldav_path):
"""Normalize CalDAV path to consistent format."""
if not caldav_path.startswith("/"):
caldav_path = "/" + caldav_path
if not caldav_path.endswith("/"):
caldav_path = caldav_path + "/"
return caldav_path
def create(self, request):
"""
Create or get existing subscription token.
POST body:
- caldav_path: The CalDAV path (e.g., /calendars/user@example.com/uuid/)
- calendar_name: Display name of the calendar (optional)
"""
create_serializer = serializers.CalendarSubscriptionTokenCreateSerializer(
data=request.data
)
create_serializer.is_valid(raise_exception=True)
caldav_path = create_serializer.validated_data["caldav_path"]
calendar_name = create_serializer.validated_data.get("calendar_name", "")
# Verify user has access to this calendar
if not self._verify_caldav_access(request.user, caldav_path):
return drf_response.Response(
{"error": "You don't have access to this calendar"},
status=status.HTTP_403_FORBIDDEN,
)
# Get or create token
token, created = models.CalendarSubscriptionToken.objects.get_or_create(
owner=request.user,
caldav_path=caldav_path,
defaults={"calendar_name": calendar_name},
)
# Update calendar_name if provided and different
if not created and calendar_name and token.calendar_name != calendar_name:
token.calendar_name = calendar_name
token.save(update_fields=["calendar_name"])
serializer = self.get_serializer(token, context={"request": request})
return drf_response.Response(
serializer.data,
status=status.HTTP_201_CREATED if created else status.HTTP_200_OK,
)
@action(detail=False, methods=["get", "delete"], url_path="by-path")
def by_path(self, request):
"""
Get or delete subscription token by CalDAV path.
Query parameter:
- caldav_path: The CalDAV path (e.g., /calendars/user@example.com/uuid/)
"""
caldav_path = request.query_params.get("caldav_path")
if not caldav_path:
return drf_response.Response(
{"error": "caldav_path query parameter is required"},
status=status.HTTP_400_BAD_REQUEST,
)
caldav_path = self._normalize_caldav_path(caldav_path)
# Verify user has access to this calendar
if not self._verify_caldav_access(request.user, caldav_path):
return drf_response.Response(
{"error": "You don't have access to this calendar"},
status=status.HTTP_403_FORBIDDEN,
)
try:
token = models.CalendarSubscriptionToken.objects.get(
owner=request.user,
caldav_path=caldav_path,
)
except models.CalendarSubscriptionToken.DoesNotExist:
return drf_response.Response(
{"error": "No subscription token exists for this calendar"},
status=status.HTTP_404_NOT_FOUND,
)
if request.method == "GET":
serializer = self.get_serializer(token, context={"request": request})
return drf_response.Response(serializer.data)
elif request.method == "DELETE":
token.delete()
return drf_response.Response(status=status.HTTP_204_NO_CONTENT)

View File

@@ -5,12 +5,15 @@ import secrets
from django.conf import settings
from django.http import HttpResponse
from django.urls import reverse
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.csrf import csrf_exempt
import requests
from core.services.calendar_invitation_service import calendar_invitation_service
logger = logging.getLogger(__name__)
@@ -31,7 +34,7 @@ class CalDAVProxyView(View):
if request.method == "OPTIONS":
response = HttpResponse(status=200)
response["Access-Control-Allow-Methods"] = (
"GET, OPTIONS, PROPFIND, REPORT, MKCOL, MKCALENDAR, PUT, DELETE"
"GET, OPTIONS, PROPFIND, PROPPATCH, REPORT, MKCOL, MKCALENDAR, PUT, DELETE, POST"
)
response["Access-Control-Allow-Headers"] = (
"Content-Type, depth, authorization, if-match, if-none-match, prefer"
@@ -81,6 +84,21 @@ class CalDAVProxyView(View):
headers["X-Api-Key"] = outbound_api_key
# Add callback URL for CalDAV scheduling (iTip/iMip)
# The CalDAV server will call this URL when it needs to send invitations
# Use CALDAV_CALLBACK_BASE_URL if configured (for Docker environments where
# the CalDAV container needs to reach Django via internal network)
callback_path = reverse("caldav-scheduling-callback")
callback_base_url = getattr(settings, "CALDAV_CALLBACK_BASE_URL", None)
if callback_base_url:
# Use configured internal URL (e.g., http://backend:8000)
headers["X-CalDAV-Callback-URL"] = (
f"{callback_base_url.rstrip('/')}{callback_path}"
)
else:
# Fall back to external URL (works when CalDAV can reach Django externally)
headers["X-CalDAV-Callback-URL"] = request.build_absolute_uri(callback_path)
# No Basic Auth - our custom backend uses X-Forwarded-User header and API key
auth = None
@@ -116,13 +134,12 @@ class CalDAVProxyView(View):
allow_redirects=False,
)
# Log authentication failures for debugging
# Log authentication failures for debugging (without sensitive headers)
if response.status_code == 401:
logger.warning(
"CalDAV server returned 401 for user %s at %s. Headers sent: %s",
"CalDAV server returned 401 for user %s at %s",
user_principal,
target_url,
headers,
)
# Build Django response
@@ -188,7 +205,13 @@ class CalDAVSchedulingCallbackView(View):
Endpoint for receiving CalDAV scheduling messages (iMip) from sabre/dav.
This endpoint receives scheduling messages (invites, responses, cancellations)
from the CalDAV server and processes them. Authentication is via API key.
from the CalDAV server and processes them by sending email notifications
with ICS attachments. Authentication is via API key.
Supported iTip methods (RFC 5546):
- REQUEST: New invitation or event update
- CANCEL: Event cancellation
- REPLY: Attendee response (accept/decline/tentative)
See: https://sabre.io/dav/scheduling/
"""
@@ -211,21 +234,80 @@ class CalDAVSchedulingCallbackView(View):
# Extract headers
sender = request.headers.get("X-CalDAV-Sender", "")
recipient = request.headers.get("X-CalDAV-Recipient", "")
method = request.headers.get("X-CalDAV-Method", "")
method = request.headers.get("X-CalDAV-Method", "").upper()
# Validate required fields
if not sender or not recipient or not method:
logger.error(
"CalDAV scheduling callback missing required headers: "
"sender=%s, recipient=%s, method=%s",
sender,
recipient,
method,
)
return HttpResponse(
status=400,
content="Missing required headers: X-CalDAV-Sender, X-CalDAV-Recipient, X-CalDAV-Method",
content_type="text/plain",
)
# Get iCalendar data from request body
icalendar_data = (
request.body.decode("utf-8", errors="replace") if request.body else ""
)
if not icalendar_data:
logger.error("CalDAV scheduling callback received empty body")
return HttpResponse(
status=400,
content="Missing iCalendar data in request body",
content_type="text/plain",
)
# For now, just log the scheduling message
logger.info(
"Received CalDAV scheduling callback: %s -> %s (method: %s)",
"Processing CalDAV scheduling message: %s -> %s (method: %s)",
sender,
recipient,
method,
)
# Log message body (first 500 chars)
if request.body:
body_preview = request.body[:500].decode("utf-8", errors="ignore")
logger.info("Scheduling message body (first 500 chars): %s", body_preview)
# Send the invitation/notification email
try:
success = calendar_invitation_service.send_invitation(
sender_email=sender,
recipient_email=recipient,
method=method,
icalendar_data=icalendar_data,
)
# TODO: Process the scheduling message (send email, update calendar, etc.)
# For now, just return success
return HttpResponse(status=200, content_type="text/plain")
if success:
logger.info(
"Successfully sent calendar %s email: %s -> %s",
method,
sender,
recipient,
)
return HttpResponse(
status=200,
content="OK",
content_type="text/plain",
)
else:
logger.error(
"Failed to send calendar %s email: %s -> %s",
method,
sender,
recipient,
)
return HttpResponse(
status=500,
content="Failed to send email",
content_type="text/plain",
)
except Exception as e:
logger.exception("Error processing CalDAV scheduling callback: %s", e)
return HttpResponse(
status=500,
content=f"Internal error: {str(e)}",
content_type="text/plain",
)

View File

@@ -0,0 +1,122 @@
"""iCal subscription export views."""
import logging
from django.conf import settings
from django.http import Http404, HttpResponse
from django.utils import timezone
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.csrf import csrf_exempt
import requests
from core.models import CalendarSubscriptionToken
logger = logging.getLogger(__name__)
@method_decorator(csrf_exempt, name="dispatch")
class ICalExportView(View):
"""
Public endpoint for iCal calendar exports.
This view serves calendar data in iCal format without requiring authentication.
The token in the URL path acts as the authentication mechanism.
URL format: /ical/<uuid:token>.ics
The view proxies the request to SabreDAV's ICSExportPlugin, which generates
RFC 5545 compliant iCal data.
"""
def get(self, request, token):
"""Handle GET requests for iCal export."""
# Lookup token
subscription = (
CalendarSubscriptionToken.objects.filter(token=token, is_active=True)
.select_related("owner")
.first()
)
if not subscription:
logger.warning("Invalid or inactive subscription token: %s", token)
raise Http404("Calendar not found")
# Update last_accessed_at atomically to avoid race conditions
# when multiple calendar clients poll simultaneously
CalendarSubscriptionToken.objects.filter(token=token, is_active=True).update(
last_accessed_at=timezone.now()
)
# Proxy to SabreDAV
caldav_url = settings.CALDAV_URL
outbound_api_key = settings.CALDAV_OUTBOUND_API_KEY
if not outbound_api_key:
logger.error("CALDAV_OUTBOUND_API_KEY is not configured")
return HttpResponse(status=500, content="iCal export not configured")
# Build the CalDAV export URL
# caldav_path is like "/calendars/user@example.com/calendar-uuid/"
# We need to call /api/v1.0/caldav/calendars/user@example.com/calendar-uuid?export
base_uri_path = "/api/v1.0/caldav"
caldav_path = subscription.caldav_path.lstrip("/")
target_url = f"{caldav_url}{base_uri_path}/{caldav_path}?export"
# Prepare headers for CalDAV server
headers = {
"X-Forwarded-User": subscription.owner.email,
"X-Api-Key": outbound_api_key,
}
try:
logger.debug(
"Proxying iCal export for caldav_path %s to %s",
subscription.caldav_path,
target_url,
)
response = requests.get(
target_url,
headers=headers,
timeout=30, # Balanced timeout: allows large calendars while preventing DoS
)
if response.status_code != 200:
logger.error(
"CalDAV server returned %d for iCal export: %s",
response.status_code,
response.content[:500],
)
return HttpResponse(
status=502,
content="Error generating calendar data",
content_type="text/plain",
)
# Return ICS response
django_response = HttpResponse(
content=response.content,
status=200,
content_type="text/calendar; charset=utf-8",
)
# Set filename for download (use calendar_name or fallback to "calendar")
display_name = subscription.calendar_name or "calendar"
safe_name = display_name.replace('"', '\\"')
django_response["Content-Disposition"] = (
f'attachment; filename="{safe_name}.ics"'
)
# Prevent caching of potentially sensitive data
django_response["Cache-Control"] = "no-store, private"
# Prevent token leakage via referrer
django_response["Referrer-Policy"] = "no-referrer"
return django_response
except requests.exceptions.RequestException as e:
logger.error("CalDAV server error during iCal export: %s", str(e))
return HttpResponse(
status=502,
content="Calendar server unavailable",
content_type="text/plain",
)

View File

@@ -8,12 +8,18 @@ from rest_framework.routers import DefaultRouter
from core.api import viewsets
from core.api.viewsets_caldav import CalDAVProxyView, CalDAVSchedulingCallbackView
from core.api.viewsets_ical import ICalExportView
from core.external_api import viewsets as external_api_viewsets
# - Main endpoints
router = DefaultRouter()
router.register("users", viewsets.UserViewSet, basename="users")
router.register("calendars", viewsets.CalendarViewSet, basename="calendars")
router.register(
"subscription-tokens",
viewsets.SubscriptionTokenViewSet,
basename="subscription-tokens",
)
urlpatterns = [
path(
@@ -41,6 +47,13 @@ urlpatterns = [
),
),
path(f"api/{settings.API_VERSION}/config/", viewsets.ConfigView.as_view()),
# Public iCal export endpoint (no authentication required)
# Token in URL acts as authentication
path(
"ical/<uuid:token>.ics",
ICalExportView.as_view(),
name="ical-export",
),
]