review fixes

Author: sylvinus

src/backend/core/services/calendar/service.py

@@ -9,7 +9,7 @@
 import logging
 import uuid
 from datetime import timezone
-from urllib.parse import urljoin
+from urllib.parse import urljoin, urlparse
 
 from django.conf import settings as django_settings
 
@@ -254,19 +254,40 @@ def _update_partstat(cal, attendee_email, new_partstat):
             if not isinstance(attendees, list):
                 attendees = [attendees]
             for att in attendees:
-                if email_lower not in str(att).lower():
+                addr = str(att).strip().lower()
+                if addr.startswith("mailto:"):
+                    addr = addr[len("mailto:") :]
+                if addr != email_lower:
                     continue
                 att.params["PARTSTAT"] = new_partstat
                 att.params.pop("RSVP", None)
 
     def _pick_calendar_url(self, calendar_id):
-        if calendar_id:
-            return calendar_id
+        if calendar_id and not self._same_origin(calendar_id):
+            raise CalDAVError(
+                "Calendar URL host does not match the configured CalDAV server."
+            )
         calendars = self.list_calendars()
         if not calendars:
             raise CalDAVError("No calendars available on this CalDAV server.")
+        if calendar_id:
+            valid_ids = {c["id"] for c in calendars}
+            if calendar_id not in valid_ids:
+                raise CalDAVError("Calendar is not in this user's calendar list.")
+            return calendar_id
         return calendars[0]["id"]
 
+    def _same_origin(self, candidate_url):
+        """Whether ``candidate_url`` shares scheme + host + port with ``self.url``."""
+        cand = urlparse(candidate_url)
+        base = urlparse(self.url)
+        if not cand.scheme or not cand.netloc:
+            return False
+        return (cand.scheme.lower(), cand.netloc.lower()) == (
+            base.scheme.lower(),
+            base.netloc.lower(),
+        )
+
     def _put_event(self, calendar_url, ics_data):
         uid = ""
         try:

src/backend/core/tests/api/test_calendar.py

@@ -16,7 +16,7 @@
 
 from core import factories
 from core.enums import MailboxRoleChoices
-from core.services.calendar.service import CalDAVService
+from core.services.calendar.service import CalDAVError, CalDAVService
 
 
 class _SilentHandler(WSGIRequestHandler):
@@ -726,6 +726,93 @@ def test_leaves_other_attendees_untouched(self):
         assert by_email["mailto:me@example.com"].params["PARTSTAT"] == "ACCEPTED"
         assert by_email["mailto:other@example.com"].params["PARTSTAT"] == "NEEDS-ACTION"
 
+    def test_substring_email_does_not_match(self):
+        """An attendee whose address contains the target as a substring must
+        not be updated; only an exact email match is."""
+        ics = (
+            "BEGIN:VCALENDAR\r\n"
+            "VERSION:2.0\r\n"
+            "PRODID:-//Test//Test//EN\r\n"
+            "BEGIN:VEVENT\r\n"
+            "UID:x@example.com\r\n"
+            "DTSTART:20260101T120000Z\r\n"
+            "DTEND:20260101T130000Z\r\n"
+            "SUMMARY:X\r\n"
+            "ATTENDEE;PARTSTAT=NEEDS-ACTION:mailto:notme@example.com\r\n"
+            "ATTENDEE;PARTSTAT=NEEDS-ACTION;RSVP=TRUE:mailto:me@example.com\r\n"
+            "END:VEVENT\r\n"
+            "END:VCALENDAR\r\n"
+        )
+        cal = ICalendar.from_ical(ics)
+        CalDAVService._update_partstat(cal, "me@example.com", "ACCEPTED")
+
+        attendees = cal.walk("VEVENT")[0].get("ATTENDEE")
+        by_email = {str(a).lower(): a for a in attendees}
+        assert by_email["mailto:me@example.com"].params["PARTSTAT"] == "ACCEPTED"
+        assert by_email["mailto:notme@example.com"].params["PARTSTAT"] == "NEEDS-ACTION"
+
+
+class TestPickCalendarUrl:
+    """Direct tests for calendar URL selection / SSRF guard."""
+
+    def _service_with_calendars(self, calendar_ids):
+        service = CalDAVService(url="https://caldav.example.com/")
+        service.list_calendars = lambda: [  # type: ignore[method-assign]
+            {"id": cid, "name": cid} for cid in calendar_ids
+        ]
+        return service
+
+    def test_returns_first_when_no_id_given(self):
+        service = self._service_with_calendars(
+            ["https://caldav.example.com/u/cal1/", "https://caldav.example.com/u/cal2/"]
+        )
+        assert service._pick_calendar_url(None) == "https://caldav.example.com/u/cal1/"
+
+    def test_accepts_known_calendar_id(self):
+        service = self._service_with_calendars(
+            ["https://caldav.example.com/u/cal1/", "https://caldav.example.com/u/cal2/"]
+        )
+        assert (
+            service._pick_calendar_url("https://caldav.example.com/u/cal2/")
+            == "https://caldav.example.com/u/cal2/"
+        )
+
+    def test_rejects_arbitrary_url(self):
+        """An attacker-controlled URL must not be used as a calendar target."""
+        service = self._service_with_calendars(["https://caldav.example.com/u/cal1/"])
+        with pytest.raises(CalDAVError):
+            service._pick_calendar_url("https://attacker.example.org/evil/")
+
+    def test_rejects_cross_origin_before_listing(self):
+        """The origin check must reject foreign hosts even if list_calendars()
+        somehow returned a matching entry — defense in depth."""
+        service = CalDAVService(url="https://caldav.example.com/")
+        called = {"n": 0}
+
+        def _spy():
+            called["n"] += 1
+            return [{"id": "https://attacker.example.org/evil/", "name": "evil"}]
+
+        service.list_calendars = _spy  # type: ignore[method-assign]
+        with pytest.raises(CalDAVError):
+            service._pick_calendar_url("https://attacker.example.org/evil/")
+        assert called["n"] == 0
+
+    def test_rejects_scheme_relative_or_malformed(self):
+        service = self._service_with_calendars(["https://caldav.example.com/u/cal1/"])
+        with pytest.raises(CalDAVError):
+            service._pick_calendar_url("/u/cal1/")
+
+    def test_rejects_unknown_calendar_on_same_host(self):
+        service = self._service_with_calendars(["https://caldav.example.com/u/cal1/"])
+        with pytest.raises(CalDAVError):
+            service._pick_calendar_url("https://caldav.example.com/u/other/")
+
+    def test_raises_when_no_calendars(self):
+        service = self._service_with_calendars([])
+        with pytest.raises(CalDAVError):
+            service._pick_calendar_url(None)
+
 
 # ---------------------------------------------------------------------------
 # Credential contract: Basic Auth user = mailbox email, password = setting