diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index cc2f19ad963e92513d157856c32279b5c5b5cdf5..66ae876f1e2d31f45635ed1409c0ebd54b7d4d47 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -65,6 +65,7 @@ Added
* Introduce .well-known urlpatterns for apps
* Global school term select for limiting data to a specific school term.
* [Dev] Notifications based on calendar alarms.
+* CalDAV and CardDAV for syncing calendars and Persons read-only.
Changed
~~~~~~~
diff --git a/aleksis/core/apps.py b/aleksis/core/apps.py
index ccaa9e948707884235309cb80406662a61acdb5d..2b512384a8011b0656568e37bbe69c0e13a9db15 100644
--- a/aleksis/core/apps.py
+++ b/aleksis/core/apps.py
@@ -62,7 +62,7 @@ class CoreConfig(AppConfig):
from django.conf import settings # noqa
# Autodiscover various modules defined by AlekSIS
- autodiscover_modules("model_extensions", "form_extensions", "checks")
+ autodiscover_modules("model_extensions", "form_extensions", "checks", "util.dav_handler")
personpreferencemodel = self.get_model("PersonPreferenceModel")
grouppreferencemodel = self.get_model("GroupPreferenceModel")
diff --git a/aleksis/core/managers.py b/aleksis/core/managers.py
index fccc79daae0e5267ef7912a8ae62850dc00c2a20..8ab80c9c4d6bda706b80f5477738bae7e727ab19 100644
--- a/aleksis/core/managers.py
+++ b/aleksis/core/managers.py
@@ -47,12 +47,20 @@ class PolymorphicBaseManager(AlekSISBaseManagerWithoutMigrations, PolymorphicMan
"""Default manager for extensible, polymorphic models."""
-class RecurrencePolymorphicQuerySet(PolymorphicQuerySet, CTEQuerySet):
+class CalendarEventMixinQuerySet(CTEQuerySet):
pass
-class RecurrencePolymorphicManager(PolymorphicBaseManager, RecurrenceManager):
- queryset_class = RecurrencePolymorphicQuerySet
+class CalendarEventQuerySet(PolymorphicQuerySet, CalendarEventMixinQuerySet):
+ pass
+
+
+class CalendarEventMixinManager(RecurrenceManager):
+ queryset_class = CalendarEventMixinQuerySet
+
+
+class CalendarEventManager(PolymorphicBaseManager, CalendarEventMixinManager):
+ queryset_class = CalendarEventQuerySet
class DateRangeQuerySetMixin:
@@ -211,7 +219,7 @@ class InstalledWidgetsDashboardWidgetOrderManager(Manager):
return super().get_queryset().filter(widget_id__in=dashboard_widget_pks)
-class HolidayQuerySet(DateRangeQuerySetMixin, RecurrencePolymorphicQuerySet):
+class HolidayQuerySet(DateRangeQuerySetMixin, CalendarEventQuerySet):
"""QuerySet with custom query methods for holidays."""
def get_all_days(self) -> list[date]:
@@ -222,5 +230,5 @@ class HolidayQuerySet(DateRangeQuerySetMixin, RecurrencePolymorphicQuerySet):
return holiday_days
-class HolidayManager(RecurrencePolymorphicManager):
+class HolidayManager(CalendarEventManager):
queryset_class = HolidayQuerySet
diff --git a/aleksis/core/migrations/0072_birthdayevent_view.py b/aleksis/core/migrations/0072_birthdayevent_view.py
new file mode 100644
index 0000000000000000000000000000000000000000..f0952560f802ddc5c897c8234a9b510da5eb127e
--- /dev/null
+++ b/aleksis/core/migrations/0072_birthdayevent_view.py
@@ -0,0 +1,26 @@
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('core', '0071_constrain_calendar_event_starting_before_ending'),
+ migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+ ]
+
+ operations = [
+ migrations.RunSQL(
+ """
+ -- Create view for BirthdayEvents
+ CREATE VIEW core_birthdayevent AS
+ SELECT id,
+ id AS person_id,
+ date_of_birth AS dt_start,
+ 'YEARLY' AS rrule
+ FROM core_person
+ WHERE date_of_birth IS NOT NULL;
+ """
+ ),
+ ]
diff --git a/aleksis/core/mixins.py b/aleksis/core/mixins.py
index fcba78889865e00191e81df9cbe771ec54be8e29..38f2a970a34fa22af3fffe72f4412674293d6f1a 100644
--- a/aleksis/core/mixins.py
+++ b/aleksis/core/mixins.py
@@ -1,19 +1,23 @@
# flake8: noqa: DJ12
import os
-from collections.abc import Iterable
+import warnings
+from collections.abc import Iterable, Sequence
from datetime import datetime
+from hashlib import sha256
from typing import TYPE_CHECKING, Any, Callable, ClassVar, Optional, Union
+from xml.etree import ElementTree
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.views import LoginView, RedirectURLMixin
from django.db import models
-from django.db.models import JSONField, QuerySet
+from django.db.models import JSONField, Q, QuerySet
from django.db.models.fields import CharField, TextField
from django.forms.forms import BaseForm
from django.forms.models import ModelForm, ModelFormMetaclass, fields_for_model
from django.http import HttpRequest, HttpResponse
+from django.urls import reverse
from django.utils import timezone
from django.utils.functional import classproperty, lazy
from django.utils.translation import gettext as _
@@ -52,8 +56,8 @@ class _ExtensibleModelBase(models.base.ModelBase):
- Register all AlekSIS models with django-reversion
"""
- def __new__(mcls, name, bases, attrs):
- mcls = super().__new__(mcls, name, bases, attrs)
+ def __new__(mcls, name, bases, attrs, **kwargs):
+ mcls = super().__new__(mcls, name, bases, attrs, **kwargs)
if "Meta" not in attrs or not attrs["Meta"].abstract:
# Register all non-abstract models with django-reversion
@@ -491,10 +495,21 @@ class RegistryObject:
_registry: ClassVar[Optional[dict[str, type["RegistryObject"]]]] = None
name: ClassVar[str] = ""
- def __init_subclass__(cls):
- if getattr(cls, "_registry", None) is None:
+ def __init_subclass__(cls, is_registry=False, register=True):
+ parent_registry = getattr(cls, "_registry", None)
+ cls._is_registry = False
+
+ if parent_registry is None or is_registry:
cls._registry = {}
- else:
+ cls._is_registry = True
+
+ if getattr(cls, "_parent_registries", None) is None:
+ cls._parent_registries = []
+
+ if parent_registry is not None:
+ cls._parent_registries.append(parent_registry)
+
+ elif register:
if not cls.name:
cls.name = cls.__name__
cls._register()
@@ -504,28 +519,158 @@ class RegistryObject:
if cls.name and cls.name not in cls._registry:
cls._registry[cls.name] = cls
+ for registry in cls._parent_registries:
+ registry[cls.name] = cls
+
+ @classmethod
+ def get_registry_objects(cls) -> dict[str, type["RegistryObject"]]:
+ return cls._registry
+
+ @classmethod
+ def get_registry_objects_recursive(cls) -> dict[str, type["RegistryObject"]]:
+ objs = cls._registry.copy()
+ for sub_registry in cls.get_sub_registries().values():
+ objs |= sub_registry.get_registry_objects_recursive()
+ return objs
+
+ @classmethod
+ def get_sub_registries(cls) -> dict[str, type["RegistryObject"]]:
+ registries = {}
+ for registry in cls.__subclasses__():
+ if registry._registry:
+ registries[registry.name] = registry
+ return registries
+
+ @classmethod
+ def get_sub_registries_recursive(cls) -> dict[str, type["RegistryObject"]]:
+ registries = {}
+ for sub_registry in cls.get_sub_registries().values():
+ registries |= sub_registry.get_sub_registries()
+ return registries
+
@classproperty
def registered_objects_dict(cls) -> dict[str, type["RegistryObject"]]:
"""Get dict of registered objects."""
- return cls._registry
+ return cls.get_registry_objects()
@classproperty
def registered_objects_list(cls) -> list[type["RegistryObject"]]:
"""Get list of registered objects."""
- return list(cls._registry.values())
+ return list(cls.get_registry_objects().values())
@classmethod
def get_object_by_name(cls, name: str) -> Optional[type["RegistryObject"]]:
"""Get registered object by name."""
return cls.registered_objects_dict.get(name)
+ @classmethod
+ def get_sub_registry_by_name(cls, name: str) -> Optional[type["RegistryObject"]]:
+ return cls.get_sub_registries().get(name)
+
class ObjectAuthenticator(RegistryObject):
def authenticate(self, request, obj):
raise NotImplementedError()
-class CalendarEventMixin(RegistryObject):
+class DAVResource(RegistryObject):
+ """Mixin for objects to provide via DAV."""
+
+ dav_verbose_name: ClassVar[str] = ""
+ dav_content_type: ClassVar[str] = ""
+
+ dav_ns: ClassVar[dict[str, str]] = {}
+ dav_resource_types: ClassVar[list[str]] = []
+
+ # Hint: We do not support dead properties for now
+ dav_live_props: ClassVar[list[tuple[str, str]]] = [
+ ("DAV:", "displayname"),
+ ("DAV:", "resourcetype"),
+ ("DAV:", "getcontenttype"),
+ ("DAV:", "getcontentlength"),
+ ]
+ dav_live_prop_methods: ClassVar[dict[tuple[str, str], str]] = {}
+
+ @classmethod
+ def get_dav_verbose_name(cls, request: Optional[HttpRequest] = None) -> str:
+ """Return the verbose name of the calendar feed."""
+ return str(cls.dav_verbose_name)
+
+ @classmethod
+ def _register_dav_ns(cls):
+ for prefix, url in cls.dav_ns.items():
+ ElementTree.register_namespace(prefix, url)
+ ElementTree.register_namespace("d", "DAV:")
+
+ @classmethod
+ def _add_dav_propnames(cls, prop: ElementTree.SubElement) -> None:
+ for ns, propname in cls.dav_live_props:
+ ElementTree.SubElement(prop, f"{{{ns}}}{propname}")
+
+ @classmethod
+ def get_dav_absolute_url(cls, reference_object, request: HttpRequest) -> str:
+ raise NotImplementedError
+
+ @classmethod
+ def get_dav_file_content(
+ cls,
+ request: HttpRequest,
+ objects: Optional[Iterable | QuerySet] = None,
+ params: Optional[dict[str, any]] = None,
+ ) -> str:
+ raise NotImplementedError
+
+ @classmethod
+ def get_dav_content_type(cls) -> str:
+ return cls.dav_content_type
+
+ @classmethod
+ def getetag(cls, request: HttpRequest, objects) -> str:
+ warnings.warn(
+ f"""The class {cls.__name__} does not override the getetag method and uses the slow and potentially unstable default implementation.""" # noqa: E501
+ )
+ try:
+ content = cls.get_dav_file_content(request, objects)
+ except NotImplementedError:
+ content = b""
+ etag = sha256()
+ etag.update(content)
+ digest = etag.hexdigest()
+ return digest
+
+
+class ContactMixin(DAVResource, RegistryObject, is_registry=True):
+ name: ClassVar[str] = "contact" # Unique name for the calendar feed
+ verbose_name: ClassVar[str] = "" # Shown name of the feed
+ link: ClassVar[str] = "" # Link for the feed, optional
+ description: ClassVar[str] = "" # Description of the feed, optional
+ color: ClassVar[str] = "#222222" # Color of the feed, optional
+ permission_required: ClassVar[str] = ""
+
+ dav_ns = {
+ "carddav": "urn:ietf:params:xml:ns:carddav",
+ }
+ dav_resource_types = ["{urn:ietf:params:xml:ns:carddav}addressbook"]
+ dav_content_type = "text/vcard"
+
+ dav_live_props: ClassVar[list[tuple[str, str]]] = [
+ ("DAV:", "displayname"),
+ ("DAV:", "resourcetype"),
+ ("DAV:", "getcontenttype"),
+ ("DAV:", "getcontentlength"),
+ (dav_ns["carddav"], "addressbook-description"),
+ ]
+
+ @classmethod
+ def get_dav_absolute_url(cls, reference_object, request: HttpRequest) -> str:
+ return reverse("dav_resource_contact", args=["contact", cls.name, reference_object.id])
+
+ @classmethod
+ def value_unique_id(cls, reference_object, request: HttpRequest) -> str:
+ return f"{cls.name}-{reference_object.id}"
+
+
+class CalendarEventMixin(DAVResource, RegistryObject, is_registry=True):
"""Mixin for calendar feeds.
This mixin can be used to create calendar feeds for objects. It can be used
@@ -562,12 +707,28 @@ class CalendarEventMixin(RegistryObject):
for the attribute. The method is called with the reference object as argument.
"""
- name: str = "" # Unique name for the calendar feed
- verbose_name: str = "" # Shown name of the feed
- link: str = "" # Link for the feed, optional
- description: str = "" # Description of the feed, optional
- color: str = "#222222" # Color of the feed, optional
- permission_required: str = ""
+ name: ClassVar[str] = "calendar" # Unique name for the calendar feed
+ verbose_name: ClassVar[str] = "" # Shown name of the feed
+ link: ClassVar[str] = "" # Link for the feed, optional
+ description: ClassVar[str] = "" # Description of the feed, optional
+ color: ClassVar[str] = "#222222" # Color of the feed, optional
+ permission_required: ClassVar[str] = ""
+
+ dav_ns = {
+ "cal": "urn:ietf:params:xml:ns:caldav",
+ "ical": "http://apple.com/ns/ical/",
+ }
+ dav_resource_types = ["{urn:ietf:params:xml:ns:caldav}calendar"]
+ dav_content_type = "text/calendar; charset=utf8; component=vevent"
+
+ dav_live_props: ClassVar[list[tuple[str, str]]] = [
+ ("DAV:", "displayname"),
+ ("DAV:", "resourcetype"),
+ ("DAV:", "getcontenttype"),
+ ("DAV:", "getcontentlength"),
+ (dav_ns["ical"], "calendar-color"),
+ (dav_ns["cal"], "calendar-description"),
+ ]
@classmethod
def get_verbose_name(cls, request: Optional[HttpRequest] = None) -> str:
@@ -617,6 +778,7 @@ class CalendarEventMixin(RegistryObject):
"""Create an event for the given reference object and add it to the feed."""
values = {}
values["timestamp"] = timezone.now()
+ values["description"] = None
for field in cls.get_event_field_names():
field_value = cls.get_event_field_value(
reference_object, field, request=request, params=params
@@ -643,13 +805,42 @@ class CalendarEventMixin(RegistryObject):
@classmethod
def get_objects(
cls,
- request: Optional[HttpRequest] = None,
- params: Optional[dict[str, any]] = None,
+ request: HttpRequest | None = None,
+ params: dict[str, any] | None = None,
start: Optional[datetime] = None,
end: Optional[datetime] = None,
- ) -> Iterable:
- """Return the objects to create the calendar feed for."""
- raise NotImplementedError
+ start_qs: QuerySet | None = None,
+ additional_filter: Q | None = None,
+ select_related: Sequence | None = None,
+ prefetch_related: Sequence | None = None,
+ expand_start: Optional[datetime] = None,
+ expand_end: Optional[datetime] = None,
+ expand: bool = False,
+ ) -> QuerySet:
+ """Return all objects that should be included in the calendar."""
+ qs = cls.objects if start_qs is None else start_qs
+ if isinstance(qs, PolymorphicBaseManager):
+ qs = qs.instance_of(cls)
+ if not start or not end:
+ if additional_filter is not None:
+ qs = qs.filter(additional_filter)
+ if select_related is not None:
+ qs = qs.select_related(*select_related)
+ if prefetch_related is not None:
+ qs = qs.prefetch_related(*prefetch_related)
+ else:
+ qs = cls.objects.with_occurrences(
+ start,
+ end,
+ start_qs=qs,
+ additional_filter=additional_filter,
+ select_related=select_related,
+ prefetch_related=prefetch_related,
+ expand_start=expand_start,
+ expand_end=expand_end,
+ expand=expand,
+ )
+ return qs
@classmethod
def create_feed(
@@ -658,7 +849,7 @@ class CalendarEventMixin(RegistryObject):
params: Optional[dict[str, any]] = None,
start: Optional[datetime] = None,
end: Optional[datetime] = None,
- queryset: Optional[QuerySet] = None,
+ queryset: Optional[Iterable | QuerySet] = None,
**kwargs,
) -> ExtendedICal20Feed:
"""Create the calendar feed with all events."""
@@ -685,7 +876,7 @@ class CalendarEventMixin(RegistryObject):
) -> Calendar:
"""Return the calendar object."""
feed = cls.create_feed(request=request, params=params, queryset=queryset)
- return feed.get_calendar_object()
+ return feed.get_calendar_object(params=params)
@classmethod
def get_events(
@@ -703,7 +894,7 @@ class CalendarEventMixin(RegistryObject):
feed = cls.create_feed(
request=request, params=params, start=start, end=end, queryset=queryset, **kwargs
)
- return feed.get_calendar_object(with_reference_object=with_reference_object)
+ return feed.get_calendar_object(with_reference_object=with_reference_object, params=params)
@classmethod
def get_single_events(
@@ -777,3 +968,23 @@ class CalendarEventMixin(RegistryObject):
@classmethod
def get_enabled_feeds(cls, request: HttpRequest | None = None):
return [feed for feed in cls.valid_feeds if feed.get_enabled(request)]
+
+ @classmethod
+ def get_dav_verbose_name(cls, request: Optional[HttpRequest] = None) -> str:
+ return str(cls.get_verbose_name())
+
+ @classmethod
+ def get_dav_file_content(
+ cls,
+ request: HttpRequest,
+ objects: Optional[Iterable | QuerySet] = None,
+ params: Optional[dict[str, any]] = None,
+ expand_start: datetime | None = None,
+ expand_end: datetime | None = None,
+ ):
+ feed = cls.create_feed(request, queryset=objects, params=params)
+ return feed.to_ical(params=params)
+
+ @classmethod
+ def get_dav_absolute_url(cls, reference_object, request: HttpRequest) -> str:
+ return reverse("dav_resource_calendar", args=["calendar", cls.name, reference_object.id])
diff --git a/aleksis/core/models.py b/aleksis/core/models.py
index f1b5ca2cb1a04e1bf8b36ad8aa500044070ab1ad..4f20856422c281e1954ff5364bea7aec751335da 100644
--- a/aleksis/core/models.py
+++ b/aleksis/core/models.py
@@ -62,18 +62,20 @@ from aleksis.core.data_checks import (
from .managers import (
AlekSISBaseManagerWithoutMigrations,
+ CalendarEventManager,
+ CalendarEventMixinManager,
GroupManager,
GroupQuerySet,
HolidayManager,
InstalledWidgetsDashboardWidgetOrderManager,
PersonManager,
PersonQuerySet,
- RecurrencePolymorphicManager,
SchoolTermQuerySet,
UninstallRenitentPolymorphicManager,
)
from .mixins import (
CalendarEventMixin,
+ ContactMixin,
ExtensibleModel,
ExtensiblePolymorphicModel,
GlobalPermissionModel,
@@ -172,13 +174,15 @@ class SchoolTerm(ExtensibleModel):
ordering = ["-date_start"]
-class Person(ExtensibleModel):
+class Person(ContactMixin, ExtensibleModel):
"""Person model.
A model describing any person related to a school, including, but not
limited to, students, teachers and guardians (parents).
"""
+ name = "person"
+
objects = PersonManager.from_queryset(PersonQuerySet)()
class Meta:
@@ -203,6 +207,7 @@ class Person(ExtensibleModel):
icon_ = "account-outline"
SEX_CHOICES = [("f", _("female")), ("m", _("male")), ("x", _("other"))]
+ SEX_CHOICES_VCARD = {"f": "F", "m": "M", "x": "O"}
user = models.OneToOneField(
get_user_model(),
@@ -384,6 +389,114 @@ class Person(ExtensibleModel):
vcal.params["ROLE"] = vText(role)
return vcal
+ def _is_unrequested_prop(self, efield, params):
+ """Return True if specific fields are requested and efield is not one of those.
+ If no fields are specified or efield is requested, return False"""
+
+ comp_name = "VCARD"
+
+ return (
+ params is not None
+ and comp_name in params
+ and efield is not None
+ and efield.upper() not in params[comp_name]
+ )
+
+ def as_vcard(self, request, params) -> str:
+ """Get this person as vCard.
+
+ Uses old version of vCard (3.0) thanks to chaotic breakage in Evolution.
+ """
+
+ card = [
+ "BEGIN:VCARD",
+ "VERSION:3.0",
+ "KIND:individual",
+ "PRODID:-//AlekSIS//AlekSIS//EN",
+ ]
+
+ if not self._is_unrequested_prop("UID", params):
+ # FIXME replace with UUID once implemented
+ card.append(f"UID:{request.build_absolute_uri(self.get_absolute_url())}")
+
+ # Name
+ if not self._is_unrequested_prop("FN", params):
+ card.append(f"FN:{self.addressing_name}")
+
+ if not self._is_unrequested_prop("N", params):
+ card.append(f"N:{self.last_name};{self.first_name};{self.additional_name};;")
+
+ # Birthday
+ if (
+ not self._is_unrequested_prop("BDAY", params)
+ and self.date_of_birth
+ and request.user.has_perm("core.view_personal_details", self)
+ ):
+ card.append(f"BDAY:{self.date_of_birth.isoformat()}")
+
+ # Email
+ if (
+ not self._is_unrequested_prop("EMAIL", params)
+ and self.email
+ and request.user.has_perm("core.view_contact_details", self)
+ ):
+ card.append(f"EMAIL:{self.email}")
+
+ # Phone Numbers
+ if not self._is_unrequested_prop("TEL", params):
+ if self.phone_number and request.user.has_perm("core.view_contact_details", self):
+ card.append(f"TEL;TYPE=home:{self.phone_number}")
+
+ if self.mobile_number and request.user.has_perm("core.view_contact_details", self):
+ card.append(f"TEL;TYPE=cell:{self.mobile_number}")
+
+ # Address
+ if (
+ not self._is_unrequested_prop("ADR", params)
+ and self.street
+ and self.postal_code
+ and self.place
+ and request.user.has_perm("core.view_address_rule", self)
+ ):
+ card.append(
+ f"ADR;TYPE=home:;;{self.street} {self.housenumber};"
+ f"{self.place};;{self.postal_code};"
+ )
+
+ card.append("END:VCARD")
+
+ return "\r\n".join(card) + "\r\n"
+
+ @classmethod
+ def get_objects(
+ cls,
+ request: HttpRequest | None = None,
+ start_qs: QuerySet | None = None,
+ additional_filter: Q | None = None,
+ ) -> QuerySet:
+ """Return all objects that should be included in the contact list."""
+ qs = cls.objects.all() if start_qs is None else start_qs
+ if request:
+ qs = qs.filter(
+ Q(pk=request.user.person.pk)
+ | Q(pk__in=get_objects_for_user(request.user, "core.view_personal_details", qs))
+ )
+ return qs.filter(additional_filter) if additional_filter else qs
+
+ @classmethod
+ def get_dav_file_content(
+ cls,
+ request: HttpRequest,
+ objects: Optional[Iterable | QuerySet] = None,
+ params: Optional[dict[str, any]] = None,
+ ) -> str:
+ if objects is None:
+ objects = cls.get_objects(request)
+ content = ""
+ for person in objects:
+ content += person.as_vcard(request, params)
+ return content.encode()
+
def save(self, *args, **kwargs):
# Determine all fields that were changed since last load
changed = self.user_info_tracker.changed()
@@ -1496,7 +1609,9 @@ class Room(ExtensibleModel):
]
-class CalendarEvent(CalendarEventMixin, ExtensiblePolymorphicModel, RecurrenceModel):
+class CalendarEvent(
+ CalendarEventMixin, ExtensiblePolymorphicModel, RecurrenceModel, register=False
+):
"""A planned event in a calendar.
To make use of this model, you need to inherit from this model.
@@ -1512,7 +1627,7 @@ class CalendarEvent(CalendarEventMixin, ExtensiblePolymorphicModel, RecurrenceMo
Please refer to the documentation of `CalendarEventMixin` for more information.
"""
- objects = RecurrencePolymorphicManager()
+ objects = CalendarEventManager()
datetime_start = models.DateTimeField(
verbose_name=_("Start date and time"), null=True, blank=True
@@ -1642,34 +1757,6 @@ class CalendarEvent(CalendarEventMixin, ExtensiblePolymorphicModel, RecurrenceMo
calendar_alarm.get_alarm(request) for calendar_alarm in reference_object.alarms.all()
]
- @classmethod
- def get_objects(
- cls,
- request: HttpRequest | None = None,
- params: dict[str, any] | None = None,
- start: Optional[datetime] = None,
- end: Optional[datetime] = None,
- start_qs: QuerySet | None = None,
- additional_filter: Q | None = None,
- select_related: Sequence | None = None,
- prefetch_related: Sequence | None = None,
- ) -> QuerySet:
- """Return all objects that should be included in the calendar."""
- start_qs = cls.objects if start_qs is None else start_qs
- start_qs = start_qs.instance_of(cls)
- if not start or not end:
- start = timezone.now() - timedelta(days=50)
- end = timezone.now() + timedelta(days=50)
- qs = cls.objects.with_occurrences(
- start,
- end,
- start_qs=start_qs,
- additional_filter=additional_filter,
- select_related=select_related,
- prefetch_related=prefetch_related,
- )
- return qs
-
def save(self, *args, **kwargs):
if (
self.datetime_start
@@ -1741,44 +1828,76 @@ class CalendarEvent(CalendarEventMixin, ExtensiblePolymorphicModel, RecurrenceMo
class FreeBusy(CalendarEvent):
- pass
+ name = ""
+
+ @classmethod
+ def value_title(cls, reference_object: "FreeBusy", request: HttpRequest | None = None) -> str:
+ return ""
-class BirthdayEvent(CalendarEventMixin):
+class BirthdayEvent(CalendarEventMixin, models.Model):
"""A calendar feed with all birthdays."""
name = "birthdays"
verbose_name = _("Birthdays")
permission_required = "core.view_birthday_calendar"
+ person = models.ForeignKey(Person, on_delete=models.DO_NOTHING)
+
+ objects = CalendarEventMixinManager()
+
+ class Meta:
+ managed = False
+ db_table = "core_birthdayevent"
+
+ def __str__(self):
+ return self.value_title(self)
+
@classmethod
- def value_title(cls, reference_object: Person, request: HttpRequest | None = None) -> str:
- return _("{}'s birthday").format(reference_object.addressing_name)
+ def value_title(
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
+ ) -> str:
+ return _("{}'s birthday").format(reference_object.person.addressing_name)
@classmethod
- def value_description(cls, reference_object: Person, request: HttpRequest | None = None) -> str:
- return f"{reference_object.addressing_name} \
- was born on {date_format(reference_object.date_of_birth)}."
+ def value_description(
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
+ ) -> str:
+ description = f"{reference_object.person.addressing_name} "
+ description += f"was born on {date_format(reference_object.person.date_of_birth)}."
+ return description
@classmethod
def value_start_datetime(
- cls, reference_object: Person, request: HttpRequest | None = None
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
+ ) -> date:
+ return reference_object.person.date_of_birth
+
+ @classmethod
+ def value_end_datetime(
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
) -> date:
- return reference_object.date_of_birth
+ return None
@classmethod
- def value_rrule(cls, reference_object: Person, request: HttpRequest | None = None) -> vRecur:
+ def value_rrule(
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
+ ) -> vRecur:
return build_rrule_from_text("FREQ=YEARLY")
@classmethod
- def value_unique_id(cls, reference_object: Person, request: HttpRequest | None = None) -> str:
- return f"birthday-{reference_object.id}"
+ def value_unique_id(
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
+ ) -> str:
+ return f"birthday-{reference_object.person.id}"
@classmethod
- def value_meta(cls, reference_object: Person, request: HttpRequest | None = None) -> dict:
+ def value_meta(
+ cls, reference_object: "BirthdayEvent", request: HttpRequest | None = None
+ ) -> dict:
return {
- "name": reference_object.addressing_name,
- "date_of_birth": reference_object.date_of_birth.isoformat(),
+ "name": reference_object.person.addressing_name,
+ "date_of_birth": reference_object.person.date_of_birth.isoformat(),
}
@classmethod
@@ -1792,6 +1911,8 @@ class BirthdayEvent(CalendarEventMixin):
params: dict[str, any] | None = None,
start: Optional[datetime] = None,
end: Optional[datetime] = None,
+ start_qs: QuerySet | None = None,
+ additional_filter: Q | None = None,
) -> QuerySet:
qs = Person.objects.filter(date_of_birth__isnull=False)
if request:
@@ -1799,7 +1920,14 @@ class BirthdayEvent(CalendarEventMixin):
Q(pk=request.user.person.pk)
| Q(pk__in=get_objects_for_user(request.user, "core.view_personal_details", qs))
)
- return qs
+
+ q = Q(person__in=qs)
+ if additional_filter is not None:
+ q = q & additional_filter
+
+ return super().get_objects(
+ request=request, params=params, start_qs=start_qs, additional_filter=q
+ )
class Holiday(CalendarEvent):
@@ -1896,6 +2024,10 @@ class PersonalEvent(CalendarEvent):
persons = models.ManyToManyField(Person, related_name="+", blank=True)
groups = models.ManyToManyField(Group, related_name="+", blank=True)
+ @classmethod
+ def get_description(cls, request: HttpRequest | None = None) -> str:
+ return ""
+
@classmethod
def get_color(cls, request: HttpRequest | None = None) -> str:
return get_site_preferences()["calendar__personal_event_color"]
@@ -1959,18 +2091,25 @@ class PersonalEvent(CalendarEvent):
params: dict[str, any] | None = None,
start: Optional[datetime] = None,
end: Optional[datetime] = None,
+ start_qs: QuerySet | None = None,
additional_filter: Q | None = None,
**kwargs,
) -> QuerySet:
- q = additional_filter
+ q = additional_filter if additional_filter is not None else Q()
if request:
- q = (
+ q = q & (
Q(owner=request.user.person)
| Q(persons=request.user.person)
| Q(groups__members=request.user.person)
)
qs = super().get_objects(
- request=request, params=params, start=start, end=end, additional_filter=q, **kwargs
+ request=request,
+ params=params,
+ start=start,
+ end=end,
+ start_qs=start_qs,
+ additional_filter=q,
+ **kwargs,
)
return qs
diff --git a/aleksis/core/schema/calendar.py b/aleksis/core/schema/calendar.py
index d79c7c0f99274823f9b96f6a1239134e98069be1..80330b385db1a067714bd99e70d7eca297796b22 100644
--- a/aleksis/core/schema/calendar.py
+++ b/aleksis/core/schema/calendar.py
@@ -27,7 +27,7 @@ class CalendarEventType(ObjectType):
return root["SUMMARY"]
def resolve_description(root, info, **kwargs):
- return root["DESCRIPTION"]
+ return root.get("DESCRIPTION", "")
def resolve_location(root, info, **kwargs):
return root.get("LOCATION", "")
@@ -39,7 +39,7 @@ class CalendarEventType(ObjectType):
return root["DTEND"].dt
def resolve_color(root, info, **kwargs):
- return root["COLOR"]
+ return root.get("COLOR")
def resolve_uid(root, info, **kwargs):
return root["UID"]
@@ -88,7 +88,9 @@ class CalendarType(ObjectType):
return root.get_description(info.context)
def resolve_url(root, info, **kwargs):
- return info.context.build_absolute_uri(reverse("calendar_feed", args=[root.name]))
+ return info.context.build_absolute_uri(
+ reverse("calendar_feed", args=["calendar", root.name])
+ )
def resolve_color(root, info, **kwargs):
return root.get_color(info.context)
@@ -129,4 +131,4 @@ class CalendarBaseType(ObjectType):
return CalendarEventMixin.get_enabled_feeds(info.context)
def resolve_all_feeds_url(root, info, **kwargs):
- return info.context.build_absolute_uri(reverse("all_calendar_feeds"))
+ return info.context.build_absolute_uri(reverse("all_calendar_feeds", args=["calendar"]))
diff --git a/aleksis/core/tests/models/test_calendar_event.py b/aleksis/core/tests/models/test_calendar_event.py
index f26e7dcdeccdaaea29698bd1acfe1a120b800f83..e5e663601a8ab3edce70caf03e366d2b0889f1a6 100644
--- a/aleksis/core/tests/models/test_calendar_event.py
+++ b/aleksis/core/tests/models/test_calendar_event.py
@@ -1,8 +1,8 @@
from datetime import datetime, timezone
-from zoneinfo import ZoneInfo
import pytest
from recurrence import WEEKLY, Recurrence, Rule
+from zoneinfo import ZoneInfo
from aleksis.core.models import CalendarEvent
diff --git a/aleksis/core/urls.py b/aleksis/core/urls.py
index 2949d4fa0ce892caa7364f34e451c0af3985f7a5..16e290332c434115edca2870480e5ede0fca1e01 100644
--- a/aleksis/core/urls.py
+++ b/aleksis/core/urls.py
@@ -39,6 +39,16 @@ urlpatterns = [
ConnectDiscoveryInfoView.as_view(),
name="oidc_configuration",
),
+ path(
+ ".well-known/carddav/",
+ views.DAVWellKnownView.as_view(),
+ name="wellknown_carddav",
+ ),
+ path(
+ ".well-known/caldav/",
+ views.DAVWellKnownView.as_view(),
+ name="wellknown_caldav",
+ ),
path("oauth/applications/", views.TemplateView.as_view(template_name="core/vue_index.html")),
path(
"oauth/applications/register/",
@@ -76,6 +86,27 @@ urlpatterns = [
views.ObjectRepresentationView.as_view(),
name="object_representation_anonymous",
),
+ path(
+ "feeds/<str:subregistry>/<str:name>.ics", views.ICalFeedView.as_view(), name="calendar_feed"
+ ),
+ path("feeds/<str:name>.ics", views.ICalAllFeedsView.as_view(), name="all_calendar_feeds"),
+ path("dav/", views.DAVResourceView.as_view(), name="dav_registry"),
+ path("dav/<str:name>/", views.DAVResourceView.as_view(), name="dav_subregistry"),
+ path(
+ "dav/<str:subregistry>/<str:name>/",
+ views.DAVResourceView.as_view(),
+ name="dav_resource",
+ ),
+ path(
+ "dav/<str:subregistry>/<str:name>/<int:id>.ics",
+ views.DAVSingleResourceView.as_view(),
+ name="dav_resource_calendar",
+ ),
+ path(
+ "dav/<str:subregistry>/<str:name>/<int:id>.vcf",
+ views.DAVSingleResourceView.as_view(),
+ name="dav_resource_contact",
+ ),
path("", include("django_prometheus.urls")),
path(
"django/",
@@ -387,8 +418,6 @@ urlpatterns = [
views.AssignPermissionView.as_view(),
name="assign_permission",
),
- path("feeds/<str:name>.ics", views.ICalFeedView.as_view(), name="calendar_feed"),
- path("feeds.ics", views.ICalAllFeedsView.as_view(), name="all_calendar_feeds"),
]
),
),
diff --git a/aleksis/core/util/auth_helpers.py b/aleksis/core/util/auth_helpers.py
index eaeb0030c0b9040b28d719b96af2d02f80aa7952..5c76860964077ae6fa588349084979c6d9b5e711 100644
--- a/aleksis/core/util/auth_helpers.py
+++ b/aleksis/core/util/auth_helpers.py
@@ -1,11 +1,13 @@
"""Helpers/overrides for django-allauth."""
+from base64 import b64decode
from typing import Any, Optional
+from django.contrib.auth import authenticate
from django.contrib.auth.validators import ASCIIUsernameValidator
from django.core.exceptions import ValidationError
from django.core.validators import RegexValidator
-from django.http import HttpRequest
+from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
from django.utils.translation import gettext_lazy as _
from oauth2_provider.models import AbstractApplication
@@ -124,6 +126,49 @@ class ClientProtectedResourceMixin(_ClientProtectedResourceMixin):
return required_scopes.issubset(allowed_scopes)
+class BasicAuthMixin:
+ """Mixin for protecting views using HTTP Basic Auth.
+
+ Useful for views being used outside a regular session authenticated, e.g. a
+ CalDAV client providing only username/password authentication. For its
+ `dispatch` method to be called, the BasicAuthMixin has to be the first
+ parent class. Make sure to call `super().dispatch(self, *args, **kwargs)`
+ if you implement a custom dispatch method.
+ """
+
+ def dispatch(self, request, *args, **kwargs):
+ if request.user is not None and request.user.is_authenticated:
+ return super().dispatch(request, *args, **kwargs)
+
+ auth_header = request.headers.get("Authorization")
+
+ if auth_header is None or not auth_header.startswith("Basic "):
+ return HttpResponse(
+ "Unauthorized",
+ status=401,
+ headers={"WWW-Authenticate": 'Basic realm="AlekSIS", charset="utf-8"'},
+ )
+
+ auth_data = auth_header.removeprefix("Basic ")
+ try:
+ auth_data_decoded = b64decode(auth_data).decode("ascii")
+ except UnicodeDecodeError:
+ try:
+ auth_data_decoded = b64decode(auth_data).decode("utf-8")
+ except UnicodeDecodeError:
+ return HttpResponseBadRequest(
+ "HTTP Basic Auth credentials must be encoded in UTF-8 or ASCII"
+ )
+ username, password = auth_data_decoded.split(":", 1)
+ user = authenticate(request, username=username, password=password)
+
+ if user is not None:
+ request.user = user
+ return super().dispatch(request, *args, **kwargs)
+ else:
+ return HttpResponse("Unauthorized", status=401)
+
+
def validate_username_preference_regex(value: str):
regex = get_site_preferences()["auth__allowed_username_regex"]
return RegexValidator(regex)(value)
diff --git a/aleksis/core/util/core_helpers.py b/aleksis/core/util/core_helpers.py
index 3a6aa841a4d26823e24c18a34b6e9ca50a4bfff7..7b3921f63af581d071f076f69e4f32ce7d575c23 100644
--- a/aleksis/core/util/core_helpers.py
+++ b/aleksis/core/util/core_helpers.py
@@ -24,7 +24,7 @@ from django.utils.module_loading import import_string
import django_ical.feedgenerator as feedgenerator
import recurring_ical_events
from cache_memoize import cache_memoize
-from icalendar import Calendar, Event, Todo
+from icalendar import Calendar, Component, Event, Timezone, Todo
if TYPE_CHECKING:
from django.contrib.contenttypes.models import ContentType
@@ -490,37 +490,92 @@ class ExtendedICal20Feed(feedgenerator.ICal20Feed):
Adds a method to return the actual calendar object.
"""
- def get_calendar_object(self, with_meta: bool = True, with_reference_object: bool = False):
+ def _is_unrequested_prop(
+ self, element: Component, efield: str, params: Optional[dict[str, any]] = None
+ ):
+ """Return True if specific fields are requested and efield is not one of those.
+ If no fields are specified or efield is requested, return False"""
+
+ return (
+ params is not None
+ and element.name in params
+ and efield is not None
+ and efield.upper() not in params[element.name]
+ )
+
+ def get_calendar_object(
+ self,
+ with_meta: bool = True,
+ with_reference_object: bool = False,
+ params: dict[str, any] = None,
+ ):
cal = Calendar()
- cal.add("version", "2.0")
- cal.add("calscale", "GREGORIAN")
- cal.add("prodid", "-//AlekSIS//AlekSIS//EN")
+ cal_props = {
+ "version": "2.0",
+ "calscale": "GREGORIAN",
+ "prodid": "-//AlekSIS//AlekSIS//EN",
+ }
+ for efield, val in cal_props.items():
+ if self._is_unrequested_prop(cal, efield, params):
+ continue
+
+ cal.add(efield, val)
for ifield, efield in EXTENDED_FEED_FIELD_MAP:
+ if self._is_unrequested_prop(cal, efield, params):
+ continue
+
val = self.feed.get(ifield)
- if val is not None:
+ if val:
cal.add(efield, val)
- self.write_items(cal, with_meta=with_meta, with_reference_object=with_reference_object)
+ self.write_items(
+ cal, with_meta=with_meta, with_reference_object=with_reference_object, params=params
+ )
+
+ if params is not None and "VTIMEZONE" in params and params["VTIMEZONE"]:
+ cal.add_missing_timezones()
return cal
- def write(self, outfile, encoding):
- cal = self.get_calendar_object(with_meta=False)
+ def to_ical(self, params: Optional[dict[str, any]] = None):
+ cal = self.get_calendar_object(with_meta=False, params=params)
to_ical = getattr(cal, "as_string", None)
if not to_ical:
to_ical = cal.to_ical
- outfile.write(to_ical())
+ return to_ical()
+
+ def write(self, outfile, params: Optional[dict[str, any]] = None):
+ cal = self.get_calendar_object(with_meta=False, params=params)
+
+ to_ical = getattr(cal, "as_string", None)
+ if not to_ical:
+ to_ical = cal.to_ical
+ outfile.write(self.to_ical())
+
+ def write_items(
+ self,
+ calendar,
+ with_meta: bool = True,
+ with_reference_object: bool = False,
+ params: dict[str, any] = None,
+ ):
+ if params is not None and "timezone" in params:
+ tz = Timezone.from_ical(params["timezone"])
+ else:
+ tz = None
- def write_items(self, calendar, with_meta: bool = True, with_reference_object: bool = False):
for item in self.items:
component_type = item.get("component_type")
element = Todo() if component_type == "todo" else Event()
for ifield, efield in EXTENDED_ITEM_ELEMENT_FIELD_MAP:
+ if self._is_unrequested_prop(element, efield, params):
+ continue
+
val = item.get(ifield)
- if val is not None:
+ if val:
if ifield == "attendee":
for list_item in val:
element.add(efield, list_item)
@@ -533,10 +588,35 @@ class ExtendedICal20Feed(feedgenerator.ICal20Feed):
elif ifield == "reference_object":
if with_reference_object:
element.add(efield, val, encode=False)
+ elif (
+ ifield == "start_datetime"
+ or ifield == "end_datetime"
+ or ifield == "timestamp"
+ or ifield == "created"
+ or ifield == "updateddate"
+ or ifield == "rdate"
+ or ifield == "exdate"
+ ):
+ if tz is not None:
+ val = val.astimezone(tz.to_tz())
+ element.add(efield, val)
else:
element.add(efield, val)
calendar.add_component(element)
+ if params is not None and ("expand_start" in params and "expand_end" in params):
+ recurrences = self.get_single_events(
+ start=params["expand_start"],
+ end=params["expand_end"],
+ )
+
+ for event in recurrences:
+ props = list(event.keys())
+ for prop in props:
+ if self._is_unrequested_prop(event, prop, params):
+ event.pop(prop)
+ calendar.add_component(event)
+
def get_single_events(self, start=None, end=None, with_reference_object: bool = False):
"""Get single event objects for this feed."""
events = recurring_ical_events.of(
diff --git a/aleksis/core/util/dav_handler/__init__.py b/aleksis/core/util/dav_handler/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..7a36d7c4bf82132a99de9fd73e365021d06169ee
--- /dev/null
+++ b/aleksis/core/util/dav_handler/__init__.py
@@ -0,0 +1 @@
+from . import calendar, contact, generic # noqa: F401
diff --git a/aleksis/core/util/dav_handler/base.py b/aleksis/core/util/dav_handler/base.py
new file mode 100644
index 0000000000000000000000000000000000000000..457365fa35d74f953fd7474cfc2fd15399ec4fec
--- /dev/null
+++ b/aleksis/core/util/dav_handler/base.py
@@ -0,0 +1,343 @@
+import warnings
+from typing import Optional
+from xml.etree import ElementTree
+from xml.sax.handler import ContentHandler, feature_namespaces
+from xml.sax.xmlreader import InputSource
+
+from django.core.exceptions import BadRequest
+from django.http import Http404, HttpRequest
+from django.urls import reverse
+
+from defusedxml.sax import make_parser
+
+from ...mixins import DAVResource, RegistryObject
+
+
+class ElementHandler(RegistryObject, is_registry=True):
+ """Abstract class serving as registry for ElementHandlers.
+
+ While parsing an XML document, for every XML element a handler exists for,
+ an instance of the respective sub class is created. By using the `children`
+ attribute, the tree is represented for later processing and creating the
+ response. Using `invisible` causes the Element not to be included in XML.
+ """
+
+ invisible: bool = False
+
+ def __init__(
+ self,
+ request: "DAVRequest",
+ parent: "ElementHandler",
+ attrs,
+ invisible: Optional[bool] = None,
+ ):
+ self.request = request
+ self.parent = parent
+ self.attrs = attrs
+ self.content = ""
+ if invisible is not None:
+ self.invisible = invisible
+ else:
+ self.invisible = self.__class__.invisible
+ self.children = []
+
+ @classmethod
+ def get_name(cls, attrs):
+ return cls.name
+
+ def continueToChild(self, child_class, attrs, invisible: Optional[bool] = None):
+ """Create instance of ElementHandler `child_class` and append to
+ children or return existing child."""
+
+ child = child_class(self.request, self, attrs, invisible=invisible)
+ self.children.append(child)
+ self.request.current_object = child
+
+ return child
+
+ def process(self, stage: str, base: "DAVMultistatus", response: "DAVResponse" = None):
+ """Build elements of the DAV multistatus response."""
+
+ previous_xml = base.current_xml
+
+ if not self.invisible:
+ xml_element = base.current_xml.find(self.name)
+ if xml_element is not None:
+ base.current_xml = xml_element
+
+ stage_method_name = f"process_{stage}"
+ if hasattr(self, stage_method_name) and callable(
+ stage_method := getattr(self, stage_method_name)
+ ):
+ try:
+ stage_method(base, response)
+ except Http404:
+ response.handle_unknown(self.name)
+
+ for child in self.children:
+ child.process(stage, base, response)
+
+ base.current_xml = previous_xml
+
+ def process_xml(self, base: "DAVMultistatus", response: "DAVResponse" = None):
+ """Add XML element representing self to multistatus XML tree."""
+
+ if not self.invisible:
+ base.current_xml = ElementTree.SubElement(base.current_xml, self.name)
+
+ @staticmethod
+ def _get_xml_sub(name):
+ """Convert a tuple like ("DAV:", "prop") to a string like "{DAV:}prop"."""
+
+ return f"{{{name[0]}}}{name[1]}"
+
+
+class DAVRequest(ElementHandler, ContentHandler):
+ """Handler that processes a DAV Request by parsing its XML tree.
+
+ Based on the `resource`, `obj` and `depth`, `resources` and `objects` to be
+ included the response are determined. Furthermore, `ElementHandler`s may
+ modify those.
+ """
+
+ depth: int | None
+
+ def __init__(
+ self, http_request: HttpRequest, resource: type[DAVResource], obj: Optional[DAVResource]
+ ):
+ super().__init__(self, None, {})
+ self._request = http_request
+
+ if depth := self._request.headers.get("Depth", "infinity"):
+ if depth not in ("0", "1", "infinity"):
+ raise BadRequest("Depth must be 0, 1 or infinity")
+ elif depth == "infinity":
+ self.depth = None
+ else:
+ self.depth = int(depth)
+
+ self.current_object = self
+
+ self.resource = resource
+ self.resources = []
+ self.objects = []
+
+ _is_registry = getattr(self.resource, "_is_registry", False)
+
+ if obj is None:
+ self.resources.append(self.resource)
+
+ if self.depth != 0 and _is_registry:
+ resources = []
+ if self.resource == DAVResource:
+ resources += self.resource.get_sub_registries().values()
+ if self.resource != DAVResource or self.depth is None:
+ if hasattr(self.resource, "valid_feeds"):
+ resources += self.resource.valid_feeds
+ else:
+ resources += self.resource.get_registry_objects().values()
+
+ for rcls in resources:
+ self.resources.append(rcls)
+
+ else:
+ self.objects.append(obj)
+
+ def parse(self):
+ """Start parsing the event-based XML parser."""
+
+ parser = make_parser()
+ parser.setFeature(feature_namespaces, True)
+ parser.setContentHandler(self)
+
+ source = InputSource()
+ source.setByteStream(self._request)
+ return parser.parse(source)
+
+ def startElementNS(self, name, qname, attrs):
+ """Handle start of a new XML element and continue to respective handler.
+
+ `ElementHandler`s may implement a `pre_handle` method to be called at
+ the start of an element.
+ """
+
+ xml_sub = self._get_xml_sub(name)
+ obj = self.get_object_by_name(xml_sub)
+ if obj is not None:
+ self.current_object.continueToChild(obj, attrs.copy())
+ if hasattr(obj, "pre_handle"):
+ self.current_object.pre_handle()
+ else:
+ child = NotImplementedObject(xml_sub)
+ self.current_object.children.append(child)
+ warnings.warn(f"DAVRequest could not parse {xml_sub}.")
+
+ def endElementNS(self, name, qname):
+ """Handle end of an XML element.
+
+ `ElementHandler`s may implement a `post_handle` method to be called at
+ the end of an element.
+ """
+
+ if self.current_object.name == self._get_xml_sub(name):
+ if hasattr(self.current_object, "post_handle"):
+ self.current_object.post_handle()
+ self.current_object = self.current_object.parent
+
+ def characters(self, content):
+ """Handle content of an XML element."""
+
+ self.current_object.content += content
+
+
+class DAVMultistatus:
+ """Base of a DAV multistatus response.
+
+ Processes children of DAVRequest `request` to build response XML.
+ """
+
+ name = "{DAV:}multistatus"
+
+ def __init__(self, request: DAVRequest):
+ self.request = request
+
+ self.request.resource._register_dav_ns()
+ self.xml_element = ElementTree.Element(self.name)
+ self.current_xml = self.xml_element
+
+ def process(self):
+ """Call process stages of all `children` of a `DAVRequest`.
+
+ There are the following stages, optionally implemented
+ by ElementHandlers using methods named `process_{stage}`:
+ `xml`: Build the response by adding an element to the XML tree.
+ """
+
+ for stage in ("xml",):
+ for child in self.request.children:
+ child.process(stage, self)
+
+
+class NotImplementedObject:
+ """Class to represent requested props that are not implemented."""
+
+ def __init__(self, xml_sub):
+ self.xml_sub = xml_sub
+
+ def process(self, stage, base, response=None):
+ if stage == "xml" and response is not None:
+ response.handle_unknown(self.xml_sub)
+
+
+class NotFoundObject:
+ """Class to represent requested objects that do not exist."""
+
+ def __init__(self, href):
+ self.href = href
+
+
+class DAVResponse:
+ """Part of a `DAVMultistatus` containing props of a single `resource` or `obj`."""
+
+ name = "{DAV:}response"
+
+ def __init__(self, base: "DAVMultistatus", resource, obj): # noqa: F821
+ self.base = base
+ self.resource = resource
+ self.obj = obj
+
+ self.xml_element = ElementTree.SubElement(self.base.current_xml, self.name)
+
+ self.href = ElementTree.SubElement(self.xml_element, "{DAV:}href")
+
+ if obj is None:
+ if self.resource == DAVResource:
+ self.href.text = reverse("dav_registry")
+ elif getattr(self.resource, "_is_registry", False):
+ self.href.text = reverse("dav_subregistry", args=[self.resource.name])
+ else:
+ self.href.text = reverse(
+ "dav_resource", args=[self.resource.__bases__[0].name, self.resource.name]
+ )
+ else:
+ if isinstance(obj, NotFoundObject):
+ self.href.text = obj.href
+ else:
+ self.href.text = self.resource.get_dav_absolute_url(
+ self.obj, self.base.request._request
+ )
+
+ self.propstats = {}
+
+ if isinstance(obj, NotFoundObject):
+ status = ElementTree.SubElement(self.xml_element, "{DAV:}status")
+ status.text = "HTTP/1.1 404 Not Found"
+ else:
+ self.init_propstat(200, "HTTP/1.1 200 OK")
+
+ def init_propstat(self, code, status_text):
+ """Initialize common sub elements of a DAV response."""
+
+ propstat = ElementTree.SubElement(self.xml_element, "{DAV:}propstat")
+ status = ElementTree.SubElement(propstat, "{DAV:}status")
+ status.text = status_text
+
+ self.propstats[code] = propstat
+ return propstat
+
+ def handle_unknown(self, xml_sub):
+ """Handle element for which no `ElementHandler` exists."""
+
+ if 404 not in self.propstats:
+ propstat = self.init_propstat(404, "HTTP/1.1 404 Not Found")
+ ElementTree.SubElement(propstat, "{DAV:}prop")
+
+ propstat = self.propstats[404]
+ prop = propstat.find("{DAV:}prop")
+ ElementTree.SubElement(prop, xml_sub)
+
+
+class QueryBase:
+ """Mixin for REPORT or PROPFIND query `ElementHandler`s.
+
+ Creates `DAVResponse` objects for `resources` and `objects` of a
+ `DAVRequest` to be processed in order to build the response.
+ """
+
+ def process(self, stage: str, base):
+ """Custom process method calling `children`'s process methods for each
+ response."""
+
+ previous_xml = base.current_xml
+
+ stage_method_name = f"process_{stage}"
+ if hasattr(self, stage_method_name) and callable(
+ stage_method := getattr(self, stage_method_name)
+ ):
+ stage_method(base)
+
+ for response in self.responses:
+ if isinstance(response.obj, NotFoundObject):
+ continue
+
+ base.current_xml = response.propstats[200]
+ for child in self.children:
+ child.process(stage, base, response)
+
+ base.current_xml = previous_xml
+
+ def process_xml(self, base):
+ self.responses = []
+ for resource in base.request.resources:
+ response = DAVResponse(base, resource, None)
+ self.responses.append(response)
+
+ for obj in base.request.objects:
+ response = DAVResponse(base, base.request.resource, obj)
+ self.responses.append(response)
+
+ def post_handle(self):
+ if self.request.objects is None:
+ self.request.objects = self.request.resource.get_objects(
+ request=self.request._request, start_qs=self.request.objects
+ )
diff --git a/aleksis/core/util/dav_handler/calendar.py b/aleksis/core/util/dav_handler/calendar.py
new file mode 100644
index 0000000000000000000000000000000000000000..8f3c6220caa08466171c7b0ae5d4640db474c53a
--- /dev/null
+++ b/aleksis/core/util/dav_handler/calendar.py
@@ -0,0 +1,267 @@
+from datetime import datetime
+from xml.etree import ElementTree
+
+from django.db.models import Q
+from django.http import Http404
+from django.urls import reverse
+
+from ..core_helpers import EXTENDED_ITEM_ELEMENT_FIELD_MAP
+from .base import ElementHandler, QueryBase
+from .generic import DAVHref, DAVProp
+
+
+class TimeRangeFilter(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}time-range"
+ invisible = True
+
+ def post_handle(self):
+ report_base = next(iter(self.request.children))
+
+ for k, v in self.attrs.items():
+ if k in [(None, "start"), (None, "end")]:
+ d = datetime.fromisoformat(v)
+ report_base.get_objects_args[k[1]] = d
+
+
+class TextMatch(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}text-match"
+ invisible = True
+
+ def _matches(self, obj, field_name):
+ method_name = f"value_{field_name}"
+ if not hasattr(obj, method_name) or not callable(getattr(obj, method_name)):
+ return False
+
+ return self.content.lower() in getattr(obj, method_name)(obj) # FIXME: Collations
+
+ def post_handle(self):
+ field_name = self.parent._get_field_name(self.parent.attrs.get((None, "name")))
+ objs = [
+ obj.pk
+ for obj in filter(lambda obj: self._matches(obj, field_name), self.request.objects)
+ ]
+ q = Q(pk__in=[objs])
+
+ report_base = next(iter(self.request.children))
+
+ if "additional_filter" in report_base.get_objects_args:
+ q = report_base.get_objects_args["additional_filter"] & q
+
+ report_base.get_objects_args["additional_filter"] = q
+
+
+class PropFilter(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}prop-filter"
+ invisible = True
+
+ @staticmethod
+ def _get_field_name(ical_name):
+ fields = list(filter(lambda f: f[1] == ical_name.lower(), EXTENDED_ITEM_ELEMENT_FIELD_MAP))
+ try:
+ return fields.pop()[0]
+ except StopIteration:
+ return None
+
+
+class CalDAVProp(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}prop"
+ invisible = True
+
+ @classmethod
+ def get_name(cls, attrs):
+ name = attrs.get((None, "name"))
+ return f"{cls.name}-{name}"
+
+ def _get_calendar_data(self, parent):
+ """Helper method to find the base `CalendarData` instance."""
+
+ if isinstance(parent, CalendarData):
+ return parent
+ return self._get_calendar_data(parent.parent)
+
+ def pre_handle(self):
+ calendar_data = self._get_calendar_data(self.parent.parent)
+ comp_name = self.parent.attrs.get((None, "name"))
+ prop = self.attrs.get((None, "name"))
+ calendar_data.params.setdefault(comp_name, []).append(prop)
+
+
+class CalDAVFilter(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}filter"
+ invisible = True
+
+ def pre_handle(self):
+ self.filters = {}
+
+
+class CalDAVCompFilter(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}comp-filter"
+ invisible = True
+
+
+class CalDAVComp(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}comp"
+ invisible = True
+
+ @classmethod
+ def get_name(cls, attrs):
+ name = attrs.get((None, "name"))
+ return f"{cls.name}-{name}"
+
+ def pre_handle(self):
+ if self.attrs.get((None, "name")) == "VTIMEZONE":
+ self.parent.parent.params["VTIMEZONE"] = True
+
+
+class Expand(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}expand"
+
+ def post_handle(self):
+ for k, v in self.attrs.items():
+ if k in [(None, "start"), (None, "end")]:
+ d = datetime.fromisoformat(v)
+ self.parent.params[f"expand_{k[1]}"] = d
+
+
+class CalendarData(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}calendar-data"
+
+ def pre_handle(self):
+ self.params = {}
+
+ def post_handle(self):
+ if not self.params:
+ attrs = {(None, "name"): "VCALENDAR"}
+ vcalendar = CalDAVComp(self.request, self, attrs)
+ self.children.append(vcalendar)
+
+ self.params["VTIMEZONE"] = True
+
+ for comp_name in ("VTIMEZONE", "VEVENT"):
+ attrs = {(None, "name"): comp_name}
+ comp = CalDAVComp(self.request, self, attrs)
+ vcalendar.children.append(comp)
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ if not self.invisible:
+ if response.obj is not None:
+ objects = response.resource.objects.filter(pk=response.obj.pk)
+ else:
+ objects = []
+
+ ical = response.resource.get_dav_file_content(
+ base.request._request, objects, params=self.params
+ )
+ base.current_xml.text = ical.decode()
+
+
+class CalendarColor(ElementHandler):
+ name = "{http://apple.com/ns/ical/}calendar-color"
+
+ def process_xml(self, base, response):
+ if not hasattr(response.resource, "get_color"):
+ raise Http404
+
+ super().process_xml(base, response)
+ base.current_xml.text = response.resource.get_color()
+
+
+class CalendarDescription(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}calendar-description"
+
+ def process_xml(self, base, response):
+ if not hasattr(response.resource, "get_description"):
+ raise Http404
+
+ super().process_xml(base, response)
+ base.current_xml.text = response.resource.get_description()
+
+
+class CalendarHomeSet(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}calendar-home-set"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ href = ElementTree.SubElement(base.current_xml, "{DAV:}href")
+ href.text = reverse("dav_subregistry", args=["calendar"])
+
+
+class SupportedCalendarComponentSet(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}supported-calendar-component-set"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ comp = ElementTree.SubElement(base.current_xml, "{urn:ietf:params:xml:ns:caldav}comp")
+ comp.set("name", "VEVENT")
+
+
+class SupportedCollationSet(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}supported-collation-set"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+
+ supported_collations = [
+ "i;ascii-casemap",
+ "i;octet",
+ ]
+
+ for collation in supported_collations:
+ supported_collation = ElementTree.SubElement(
+ base.current_xml, "{urn:ietf:params:xml:ns:caldav}supported-collation"
+ )
+ supported_collation.text = collation
+
+
+class Timezone(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}timezone"
+ invisible = True
+
+
+class ReportBase(QueryBase):
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ self.get_objects_args = {
+ "request": self.request._request,
+ }
+
+ def post_handle(self):
+ super().post_handle()
+
+ self.request.objects = self.request.resource.get_objects(**self.get_objects_args)
+
+ try:
+ timezone = next(filter(lambda child: isinstance(child, Timezone), self.children))
+ except StopIteration:
+ timezone = None
+
+ if timezone is not None:
+ prop = next(filter(lambda child: isinstance(child, DAVProp), self.children))
+ calendar_data = next(
+ filter(lambda child: isinstance(child, CalendarData), prop.children)
+ )
+ calendar_data.params["timezone"] = timezone.content
+
+
+class CalendarQuery(ReportBase, ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}calendar-query"
+
+ def pre_handle(self):
+ self.request.resources = []
+
+
+class CalendarMultiget(ReportBase, ElementHandler):
+ name = "{urn:ietf:params:xml:ns:caldav}calendar-multiget"
+
+ def pre_handle(self):
+ self.request.resources = []
+ self.request.objects = []
+
+ def post_handle(self):
+ super().post_handle()
+
+ for child in self.children:
+ if child.name == DAVHref.name:
+ child.invisible = True
diff --git a/aleksis/core/util/dav_handler/contact.py b/aleksis/core/util/dav_handler/contact.py
new file mode 100644
index 0000000000000000000000000000000000000000..f6f7d84e12ef960a60a4436692aaca4e0e2cb623
--- /dev/null
+++ b/aleksis/core/util/dav_handler/contact.py
@@ -0,0 +1,99 @@
+from xml.etree import ElementTree
+
+from django.http import Http404
+from django.urls import reverse
+
+from .base import ElementHandler, QueryBase
+from .generic import DAVHref
+
+
+class CardDAVProp(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}prop"
+ invisible = True
+
+ @classmethod
+ def get_name(cls, attrs):
+ name = attrs.get((None, "name"))
+ return f"{cls.name}-{name}"
+
+ def _get_address_data(self, parent):
+ """Helper method to find the base `AddressData` instance."""
+
+ if isinstance(parent, AddressData):
+ return parent
+ return self._get_address_data(parent.parent)
+
+ def pre_handle(self):
+ address_data = self._get_address_data(self.parent)
+ comp_name = "VCARD"
+ prop = self.attrs.get((None, "name"))
+ address_data.params.setdefault(comp_name, []).append(prop)
+
+
+class AddressData(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}address-data"
+
+ def pre_handle(self):
+ self.params = {}
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ if not self.invisible:
+ objects = [response.obj] if response.obj is not None else []
+ vcf = response.resource.get_dav_file_content(
+ base.request._request, objects, params=self.params
+ )
+ base.current_xml.text = vcf.decode()
+
+
+class AddressbookDescription(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}addressbook-description"
+
+ def process_xml(self, base, response):
+ if not hasattr(response.resource, "get_description"):
+ raise Http404
+
+ response.resource.get_description()
+ base.current_xml.text = response.resource.get_description()
+ super().process_xml(base, response)
+
+
+class AddressbookHomeSet(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}addressbook-home-set"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ href = ElementTree.SubElement(base.current_xml, "{DAV:}href")
+ href.text = reverse("dav_subregistry", args=["contact"])
+
+
+class SupportedAddressData(ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}supported-address-data"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ comp = ElementTree.SubElement(
+ base.current_xml, "{urn:ietf:params:xml:ns:carddav}address-data-type"
+ )
+ comp.set("content-type", "text/vcard")
+ comp.set("version", "4.0")
+
+
+class AddressbookQuery(QueryBase, ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}addressbook-query"
+
+ def pre_handle(self):
+ self.request.resources = []
+
+
+class AddressbookMultiget(QueryBase, ElementHandler):
+ name = "{urn:ietf:params:xml:ns:carddav}addressbook-multiget"
+
+ def pre_handle(self):
+ self.request.resources = []
+ self.request.objects = []
+
+ def post_handle(self):
+ for child in self.children:
+ if child.name == DAVHref.name:
+ child.invisible = True
diff --git a/aleksis/core/util/dav_handler/generic.py b/aleksis/core/util/dav_handler/generic.py
new file mode 100644
index 0000000000000000000000000000000000000000..188d50d73c5692dfaca5dfba20307b832684ad1f
--- /dev/null
+++ b/aleksis/core/util/dav_handler/generic.py
@@ -0,0 +1,184 @@
+from xml.etree import ElementTree
+
+from django.http import Http404
+from django.urls import resolve
+
+from ...mixins import DAVResource
+from .base import DAVMultistatus, DAVResponse, ElementHandler, NotFoundObject, QueryBase
+
+
+class DAVEtag(ElementHandler):
+ name = "{DAV:}getetag"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ objects = [response.obj] if response.obj is not None else []
+ etag = response.resource.getetag(base.request._request, objects)
+ base.current_xml.text = f'"{etag}"'
+
+
+class DAVDisplayname(ElementHandler):
+ name = "{DAV:}displayname"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ base.current_xml.text = response.resource.get_dav_verbose_name()
+
+
+class DAVResourcetype(ElementHandler):
+ name = "{DAV:}resourcetype"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ resource_types = ["{DAV:}collection"]
+ if not getattr(response.resource, "_is_registry", False):
+ resource_types += response.resource.dav_resource_types
+
+ if response.obj is None:
+ for res_type_attr in resource_types:
+ ElementTree.SubElement(base.current_xml, res_type_attr)
+
+
+class DAVCurrentUserPrincipal(ElementHandler):
+ name = "{DAV:}current-user-principal"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+ href = ElementTree.SubElement(base.current_xml, "{DAV:}href")
+ person_pk = base.request._request.user.person.pk
+ href.text = f"/dav/contact/person/{person_pk}.vcf" # FIXME: Implement principals
+
+
+class DAVGetcontenttype(ElementHandler):
+ name = "{DAV:}getcontenttype"
+
+ def process_xml(self, base, response):
+ resource = response.resource
+ if response.obj is not None:
+ resource = response.obj
+ content_type = resource.get_dav_content_type()
+
+ if not content_type:
+ raise Http404
+
+ super().process_xml(base, response)
+ base.current_xml.text = content_type
+
+
+class DAVGetcontentlength(ElementHandler):
+ name = "{DAV:}getcontentlength"
+
+ def process_xml(self, base, response):
+ obj = response.obj
+ if obj is None:
+ raise Http404
+
+ super().process_xml(base, response)
+ base.current_xml.text = str(
+ len(response.resource.get_dav_file_content(base.request._request, [obj]))
+ )
+
+
+class DAVSupportedReportSet(ElementHandler):
+ name = "{DAV:}supported-report-set"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+
+ supported_reports = [
+ ("urn:ietf:params:xml:ns:caldav", "calendar-multiget"),
+ ("urn:ietf:params:xml:ns:caldav", "calendar-query"),
+ ]
+
+ for r in supported_reports:
+ supported_report = ElementTree.SubElement(base.current_xml, "{DAV:}supported-report")
+ report = ElementTree.SubElement(supported_report, "{DAV:}report")
+ ElementTree.SubElement(report, self._get_xml_sub(r))
+
+
+class DAVCurrentUserPrivilegeSet(ElementHandler):
+ name = "{DAV:}current-user-privilege-set"
+
+ def process_xml(self, base, response):
+ super().process_xml(base, response)
+
+ privileges = [
+ ("DAV:", "read"),
+ ]
+
+ for p in privileges:
+ privilege = ElementTree.SubElement(base.current_xml, "{DAV:}privilege")
+ ElementTree.SubElement(privilege, self._get_xml_sub(p))
+
+
+class DAVHref(ElementHandler):
+ name = "{DAV:}href"
+
+ def post_handle(self):
+ res = resolve(self.content)
+ name = res.kwargs.get("name")
+ pk = res.kwargs.get("id")
+
+ resource = DAVResource.registered_objects_dict[name]
+ try:
+ obj = resource.get_objects(self.request._request).get(pk=pk)
+ except resource.DoesNotExist:
+ obj = NotFoundObject(self.content)
+
+ self.request.objects = list(self.request.objects)
+ self.request.objects.append(obj)
+
+
+class DAVProp(ElementHandler):
+ name = "{DAV:}prop"
+
+
+class DAVPropname(ElementHandler):
+ name = "{DAV:}propname"
+ invisible = True
+
+ def process_xml(self, base: DAVMultistatus, response: DAVResponse = None):
+ base.current_xml = ElementTree.SubElement(base.current_xml, DAVProp.name)
+
+ response.resource._add_dav_propnames(base.current_xml)
+
+
+class DAVAllprop(ElementHandler):
+ name = "{DAV:}allprop"
+ invisible = True
+
+ def pre_handle(self):
+ for name in self.request.resource.dav_live_props:
+ self.request.startElementNS(name, None, {})
+ self.request.endElementNS(name, None)
+
+ def process(self, stage: str, base: DAVMultistatus, response: DAVResponse = None):
+ xml_element = base.current_xml.find(DAVProp.name)
+ if xml_element is not None:
+ base.current_xml = xml_element
+
+ super().process(stage, base, response)
+
+ def process_xml(self, base: DAVMultistatus, response: DAVResponse = None):
+ base.current_xml = ElementTree.SubElement(base.current_xml, DAVProp.name)
+
+
+class Propfind(QueryBase, ElementHandler):
+ name = "{DAV:}propfind"
+ invisible = True
+
+ def post_handle(self):
+ super().post_handle()
+
+ _is_registry = getattr(self.request.resource, "_is_registry", False)
+
+ if not _is_registry or self.request.depth is None:
+ for resource in filter(
+ lambda r: not getattr(r, "_is_registry", False), self.request.resources
+ ):
+ try:
+ objs = resource.get_objects(self.request._request)
+ except NotImplementedError:
+ objs = []
+
+ self.request.objects += objs
diff --git a/aleksis/core/views.py b/aleksis/core/views.py
index a356ad0df0ded55ee7bdb967ef5953bd054d8897..1d5397d82547d22292ffb8cb59c906a0766c5db7 100644
--- a/aleksis/core/views.py
+++ b/aleksis/core/views.py
@@ -1,6 +1,8 @@
from textwrap import wrap
-from typing import Any, Optional
+from typing import Any, ClassVar, Optional
from urllib.parse import urlencode, urlparse, urlunparse
+from xml.etree import ElementTree
+from xml.sax import SAXParseException
from django.apps import apps
from django.conf import settings
@@ -30,6 +32,7 @@ from django.utils.decorators import method_decorator
from django.utils.translation import get_language
from django.utils.translation import gettext_lazy as _
from django.views.decorators.cache import never_cache
+from django.views.decorators.csrf import csrf_exempt
from django.views.defaults import ERROR_500_TEMPLATE_NAME
from django.views.generic.base import TemplateView, View
from django.views.generic.detail import DetailView, SingleObjectMixin
@@ -97,7 +100,9 @@ from .mixins import (
AdvancedDeleteView,
AdvancedEditView,
CalendarEventMixin,
+ DAVResource,
ObjectAuthenticator,
+ RegistryObject,
SuccessNextMixin,
)
from .models import (
@@ -124,6 +129,7 @@ from .tables import (
UserObjectPermissionTable,
)
from .util import messages
+from .util.auth_helpers import BasicAuthMixin
from .util.celery_progress import render_progress_page
from .util.core_helpers import (
generate_random_code,
@@ -133,6 +139,7 @@ from .util.core_helpers import (
has_person,
objectgetter_optional,
)
+from .util.dav_handler.base import DAVMultistatus, DAVRequest
from .util.forms import PreferenceLayout
from .util.pdf import render_pdf
@@ -1374,19 +1381,44 @@ class ObjectRepresentationView(View):
raise PermissionDenied()
-class ICalFeedView(PermissionRequiredMixin, View):
+class RegistryObjectViewMixin:
+ """Provide single registry object by its name."""
+
+ registry_object: ClassVar[type[RegistryObject] | RegistryObject] = None
+
+ def get_object(self) -> type[RegistryObject]:
+ if not self.registry_object:
+ raise NotImplementedError("There is no registry object set.")
+
+ if "name" in self.kwargs and "subregistry" not in self.kwargs:
+ return self._get_sub_registry()
+
+ name = self.kwargs.get("name")
+ if name is None:
+ return self.registry_object
+ if name in self.registry_object.registered_objects_dict:
+ return self.registry_object.registered_objects_dict[name]
+ raise Http404
+
+ def _get_sub_registry(self) -> type[RegistryObject]:
+ name = self.kwargs.get("name")
+ if name in self.registry_object.get_sub_registries():
+ return self.registry_object.get_sub_registry_by_name(name)
+ raise Http404
+
+
+class ICalFeedView(RegistryObjectViewMixin, PermissionRequiredMixin, View):
"""View to generate an iCal feed for a calendar."""
permission_required = "core.view_calendar_feed_rule"
+ registry_object = CalendarEventMixin
def get(self, request, name, *args, **kwargs):
- if name in CalendarEventMixin.registered_objects_dict:
- calendar = CalendarEventMixin.registered_objects_dict[name]
- feed = calendar.create_feed(request)
- response = HttpResponse(content_type="text/calendar")
- feed.write(response, "utf-8")
- return response
- raise Http404
+ calendar = self.get_object()
+ feed = calendar.create_feed(request)
+ response = HttpResponse(content_type="text/calendar")
+ feed.write(response)
+ return response
class ICalAllFeedsView(PermissionRequiredMixin, View):
@@ -1398,7 +1430,90 @@ class ICalAllFeedsView(PermissionRequiredMixin, View):
response = HttpResponse(content_type="text/calendar")
for calendar in CalendarEventMixin.get_enabled_feeds(request):
feed = calendar.create_feed(request)
- feed.write(response, "utf-8")
+ feed.write(response)
+ return response
+
+
+@method_decorator(csrf_exempt, name="dispatch")
+class DAVResourceView(BasicAuthMixin, PermissionRequiredMixin, RegistryObjectViewMixin, View):
+ """View for CalDAV collections."""
+
+ registry_object = DAVResource
+ permission_required = "core.view_calendar_feed_rule"
+
+ http_method_names = View.http_method_names + ["propfind", "report"]
+ dav_compliance = []
+
+ _dav_request: ClassVar[DAVRequest]
+
+ def dispatch(self, request, *args, **kwargs):
+ res = super().dispatch(request, *args, **kwargs)
+
+ res.headers["DAV"] = ", ".join(
+ ["1", "3", "calendar-access", "addressbook"] + self.dav_compliance
+ )
+
+ return res
+
+ def options(self, request, *args, **kwargs):
+ res = super().options(request, *args, **kwargs)
+ return res
+
+ def propfind(self, request, *args, **kwargs):
+ resource = self.get_object()
+ self._dav_request = DAVRequest(request, resource, None)
+
+ try:
+ self._dav_request.parse()
+ except SAXParseException:
+ return HttpResponseBadRequest()
+
+ multistatus = DAVMultistatus(self._dav_request)
+ multistatus.process()
+
+ response = HttpResponse(
+ ElementTree.tostring(multistatus.xml_element), content_type="text/xml", status=207
+ )
+ return response
+
+ def report(self, request, *args, **kwargs):
+ return self.propfind(request, *args, **kwargs)
+
+
+class DAVSingleResourceView(DAVResourceView):
+ """View for single CalDAV resources."""
+
+ def propfind(self, request, id, *args, **kwargs): # noqa: A002
+ resource = self.get_object()
+ try:
+ objects = resource.get_objects(request).get(pk=id)
+ except resource.DoesNotExist as exc:
+ raise Http404 from exc
+
+ self._dav_request = DAVRequest(request, resource, objects)
+
+ try:
+ self._dav_request.parse()
+ except SAXParseException:
+ return HttpResponseBadRequest()
+
+ multistatus = DAVMultistatus(self._dav_request)
+ multistatus.process()
+
+ response = HttpResponse(
+ ElementTree.tostring(multistatus.xml_element), content_type="text/xml", status=207
+ )
+ return response
+
+ def get(self, request, name, id, *args, **kwargs): # noqa: A002
+ resource: DAVResource = self.get_object()
+ try:
+ obj = resource.get_objects(request).get(pk=id)
+ except resource.DoesNotExist as exc:
+ raise Http404 from exc
+
+ response = HttpResponse(content_type=resource.get_dav_content_type())
+ response.write(resource.get_dav_file_content(request, objects=[obj]))
return response
@@ -1409,3 +1524,17 @@ class CustomLogoutView(LogoutView):
def get(self, *args, **kwargs):
return self.post(*args, **kwargs)
+
+
+@method_decorator(csrf_exempt, name="dispatch")
+class DAVWellKnownView(View):
+ http_method_names = View.http_method_names + ["propfind", "report"]
+
+ def get(self, request, *args, **kwargs):
+ return redirect("dav_registry")
+
+ def propfind(self, request, *args, **kwargs):
+ return redirect("dav_registry")
+
+ def post(self, request, *args, **kwargs):
+ return redirect("dav_registry")
diff --git a/pyproject.toml b/pyproject.toml
index 64e7ef4f0d7ac85b90ea5d16674de3f63088ff83..48b4290e640c69a69a9d61cf7d52b7f3af7370d8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -133,6 +133,7 @@ tqdm = "^4.66.1"
django-pg-rrule = "^0.3.1"
libsass = "^0.23.0"
graphene-django-optimizer-reloaded = "^0.9.2"
+defusedxml = "^0.7.1"
[tool.poetry.extras]
ldap = ["django-auth-ldap"]