diff --git a/src/govoplan_calendar/backend/caldav.py b/src/govoplan_calendar/backend/caldav.py index 1a67a2d..1c31551 100644 --- a/src/govoplan_calendar/backend/caldav.py +++ b/src/govoplan_calendar/backend/caldav.py @@ -295,24 +295,13 @@ class CalDAVClient: return CalDAVWriteResult(href=href, etag=response_etag(response_headers), status=status) def object_url(self, href: str) -> str: - candidate = urllib.parse.urljoin(self.collection_url, href) + candidate = same_origin_dav_url( + self.collection_url, + href, + label="CalDAV object href", + ) collection_parts = urllib.parse.urlparse(self.collection_url) candidate_parts = urllib.parse.urlparse(candidate) - if ( - candidate_parts.username - or candidate_parts.password - or ( - candidate_parts.scheme.lower(), - candidate_parts.hostname, - candidate_parts.port, - ) - != ( - collection_parts.scheme.lower(), - collection_parts.hostname, - collection_parts.port, - ) - ): - raise CalDAVError("CalDAV object href must use the configured collection origin") collection_path = posixpath.normpath(urllib.parse.unquote(collection_parts.path)) candidate_path = posixpath.normpath(urllib.parse.unquote(candidate_parts.path)) collection_prefix = collection_path.rstrip("/") + "/" @@ -364,8 +353,14 @@ class CalDAVClient: def urllib_transport(method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: int) -> tuple[int, Mapping[str, str], bytes]: url = validate_http_url(url) try: - request = urllib.request.Request(url, data=body, headers=dict(headers), method=method) - with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated CalDAV HTTP(S) URL. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + request = urllib.request.Request( # noqa: S310 - URL is validated and origin-confined. + url, + data=body, + headers=dict(headers), + method=method, + ) + opener = urllib.request.build_opener(_SameOriginRedirectHandler(url)) + with opener.open(request, timeout=timeout) as response: # noqa: S310 - validated CalDAV URL; redirects remain on origin. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected return response.status, dict(response.headers.items()), response.read() except urllib.error.HTTPError as exc: return exc.code, dict(exc.headers.items()), exc.read() @@ -504,16 +499,72 @@ def ensure_collection_url(value: str) -> str: def validate_http_url(value: str) -> str: - parsed = urllib.parse.urlparse(value) - if parsed.scheme not in {"http", "https"} or not parsed.netloc: + parsed = urllib.parse.urlparse(value.strip()) + if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc or not parsed.hostname: raise CalDAVError("CalDAV URL must be an absolute HTTP(S) URL") if parsed.username or parsed.password: raise CalDAVError("CalDAV URL must not include embedded credentials") + if parsed.query or parsed.fragment: + raise CalDAVError("CalDAV URL must not include a query or fragment") + _url_origin(parsed) return urllib.parse.urlunparse(parsed) def absolute_dav_url(base_url: str, href: str) -> str: - return urllib.parse.urljoin(ensure_collection_url(base_url), href) + return same_origin_dav_url(base_url, href, label="CalDAV discovery href") + + +def same_origin_dav_url(base_url: str, href: str, *, label: str) -> str: + base = ensure_collection_url(base_url) + candidate = validate_http_url(urllib.parse.urljoin(base, href)) + if _url_origin(urllib.parse.urlparse(candidate)) != _url_origin(urllib.parse.urlparse(base)): + raise CalDAVError(f"{label} must use the configured collection origin") + return candidate + + +def _url_origin(parsed: urllib.parse.ParseResult) -> tuple[str, str, int]: + try: + port = parsed.port + except ValueError as exc: + raise CalDAVError("CalDAV URL has an invalid port") from exc + scheme = parsed.scheme.lower() + if port is None: + port = 443 if scheme == "https" else 80 + return scheme, (parsed.hostname or "").lower(), port + + +class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): + def __init__(self, source_url: str) -> None: + super().__init__() + self._source_origin = _url_origin(urllib.parse.urlparse(validate_http_url(source_url))) + + def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def] + del fp, msg, headers + try: + candidate = validate_http_url(newurl) + except CalDAVError: + return None + if _url_origin(urllib.parse.urlparse(candidate)) != self._source_origin: + return None + method = req.get_method() + data = req.data + if code == 303 and method != "HEAD": + method, data = "GET", None + elif code in {301, 302} and method == "POST": + method, data = "GET", None + forwarded_headers = { + key: value + for key, value in req.header_items() + if key.casefold() not in {"host", "content-length"} + } + return urllib.request.Request( # noqa: S310 - candidate is validated and same-origin. + candidate, + data=data, + headers=forwarded_headers, + origin_req_host=req.origin_req_host, + unverifiable=True, + method=method, + ) def strip_weak_etag(value: str | None) -> str | None: diff --git a/src/govoplan_calendar/backend/service.py b/src/govoplan_calendar/backend/service.py index ae30d76..8b66903 100644 --- a/src/govoplan_calendar/backend/service.py +++ b/src/govoplan_calendar/backend/service.py @@ -158,14 +158,57 @@ def normalize_sync_source_url(source_kind: str, value: str) -> str: def validate_http_url(url: str, *, label: str) -> str: - parsed = urllib.parse.urlparse(url) - if parsed.scheme not in {"http", "https"} or not parsed.netloc: + parsed = urllib.parse.urlparse(url.strip()) + if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc or not parsed.hostname: raise CalendarError(f"{label} must be an absolute HTTP(S) URL") if parsed.username or parsed.password: raise CalendarError(f"{label} must not include embedded credentials") + _parsed_http_origin(parsed, label=label) return urllib.parse.urlunparse(parsed) +def same_origin_http_url(base_url: str, candidate_url: str, *, label: str) -> str: + base = validate_http_url(base_url, label=label) + candidate = validate_http_url(candidate_url, label=label) + if _parsed_http_origin(urllib.parse.urlparse(candidate), label=label) != _parsed_http_origin( + urllib.parse.urlparse(base), + label=label, + ): + raise CalendarError(f"{label} must remain on the configured source origin") + return candidate + + +def _parsed_http_origin(parsed: urllib.parse.ParseResult, *, label: str) -> tuple[str, str, int]: + try: + port = parsed.port + except ValueError as exc: + raise CalendarError(f"{label} has an invalid port") from exc + scheme = parsed.scheme.lower() + if port is None: + port = 443 if scheme == "https" else 80 + return scheme, (parsed.hostname or "").lower(), port + + +class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): + def __init__(self, source_url: str) -> None: + super().__init__() + parsed = urllib.parse.urlparse(validate_http_url(source_url, label="Calendar source URL")) + self._source_origin = _parsed_http_origin(parsed, label="Calendar source URL") + + def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def] + try: + candidate = validate_http_url(newurl, label="Calendar source redirect URL") + candidate_origin = _parsed_http_origin( + urllib.parse.urlparse(candidate), + label="Calendar source redirect URL", + ) + except CalendarError: + return None + if candidate_origin != self._source_origin: + return None + return super().redirect_request(req, fp, code, msg, headers, candidate) + + def normalize_graph_collection_url(value: str) -> str: url = value.strip() if url.startswith("/"): @@ -1186,12 +1229,28 @@ def http_request( headers: dict[str, str] | None = None, body: str | bytes | None = None, timeout: int = 30, + credential_origin: str | None = None, ) -> tuple[int, dict[str, str], str]: data = body.encode("utf-8") if isinstance(body, str) else body url = validate_http_url(url, label="Calendar source URL") - request = urllib.request.Request(url, data=data, method=method, headers=headers or {}) + credential_origin = validate_http_url( + credential_origin or url, + label="Calendar credential origin", + ) + url = same_origin_http_url( + credential_origin, + url, + label="Calendar source request URL", + ) + request = urllib.request.Request( # noqa: S310 - URL is validated and origin-confined. + url, + data=data, + method=method, + headers=headers or {}, + ) + opener = urllib.request.build_opener(_SameOriginRedirectHandler(credential_origin)) try: - with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated admin-configured sync source URL. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + with opener.open(request, timeout=timeout) as response: # noqa: S310 - validated, origin-confined Calendar URL. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected response_headers = {key.lower(): value for key, value in response.headers.items()} payload = response.read().decode(response_headers.get("content-charset") or "utf-8", errors="replace") return int(response.status), response_headers, payload @@ -1478,7 +1537,16 @@ def sync_graph_source( seen_hrefs: set[str] = set() delta_link: str | None = None for _page in range(50): - _status, _headers, body = http_request(next_url, headers=headers) + next_url = same_origin_http_url( + source.collection_url, + next_url, + label="Microsoft Graph continuation URL", + ) + _status, _headers, body = http_request( + next_url, + headers=headers, + credential_origin=source.collection_url, + ) payload = json.loads(body or "{}") for item in payload.get("value", []): href = str(item.get("id") or "") @@ -1493,10 +1561,20 @@ def sync_graph_source( stats.updated += updated stats.unchanged += unchanged next_link = payload.get("@odata.nextLink") - delta_link = payload.get("@odata.deltaLink") or delta_link + raw_delta_link = payload.get("@odata.deltaLink") + if raw_delta_link: + delta_link = same_origin_http_url( + source.collection_url, + str(raw_delta_link), + label="Microsoft Graph delta URL", + ) if not next_link: break - next_url = str(next_link) + next_url = same_origin_http_url( + source.collection_url, + str(next_link), + label="Microsoft Graph continuation URL", + ) else: raise CalendarError("Microsoft Graph sync returned too many pages") if stats.full_sync: diff --git a/tests/test_caldav.py b/tests/test_caldav.py index d96e775..ffd1e01 100644 --- a/tests/test_caldav.py +++ b/tests/test_caldav.py @@ -246,12 +246,21 @@ class CalDAVSyncTests(unittest.TestCase): create_scope_tables(self.engine) Base.metadata.create_all(bind=self.engine) self.Session = sessionmaker(bind=self.engine) + self.sessions = [] def tearDown(self) -> None: + for session in reversed(self.sessions): + session.close() Base.metadata.drop_all(bind=self.engine) + self.engine.dispose() + + def session(self): + session = self.Session() + self.sessions.append(session) + return session def test_source_creation_encrypts_credential_and_resolves_client_secret(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) @@ -278,7 +287,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual(client.password, "secret") def test_delete_calendar_retires_caldav_source_and_credential(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) source = create_caldav_source( @@ -304,7 +313,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual(list_caldav_sources(session, tenant_id="tenant-1"), []) def test_create_source_retires_orphaned_source_for_deleted_calendar(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) old_calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Old")) old_source = create_caldav_source( @@ -330,7 +339,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual([source.id for source in list_caldav_sources(session, tenant_id="tenant-1")], [new_source.id]) def test_create_source_reports_active_duplicate_as_calendar_error(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) first = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="First")) second = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Second")) @@ -351,7 +360,7 @@ class CalDAVSyncTests(unittest.TestCase): ) def test_due_sync_runs_due_sources_and_reschedules(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) source = create_caldav_source( @@ -378,7 +387,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertGreater(next_sync_at, datetime(2026, 7, 7, 8, 1, tzinfo=timezone.utc)) def test_due_sync_emits_recovery_notification_after_previous_error(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) source = create_caldav_source( @@ -406,7 +415,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual(provider.requests[0].event_kind, "calendar.sync.ok") def test_create_and_update_event_coalesce_to_committed_outbox_state(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal")) @@ -441,7 +450,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertIn("SUMMARY:Updated", event.raw_ics or "") def test_update_event_reports_remote_etag_conflict(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal")) @@ -467,7 +476,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertIn("changed remotely", operation.last_error or "") def test_delete_one_component_puts_remaining_resource_instead_of_deleting_object(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal")) @@ -490,7 +499,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertIsNotNone(override.deleted_at) def test_freebusy_expands_recurring_events_and_skips_transparent_items(self) -> None: - session = self.Session() + session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Local")) create_event( @@ -532,7 +541,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual(busy[0]["start_at"], datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc)) def test_full_sync_imports_multi_vevent_resource_and_fetches_missing_calendar_data(self) -> None: - session = self.Session() + session = self.session() tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant") session.add(tenant) calendar = create_calendar( @@ -573,7 +582,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual(calendar.metadata_["source_kind"], "caldav") def test_sync_token_deletion_soft_deletes_remote_resource_events(self) -> None: - session = self.Session() + session = self.session() tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant") session.add(tenant) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) @@ -622,7 +631,7 @@ class CalDAVSyncTests(unittest.TestCase): self.assertIsNotNone(event.deleted_at) def test_missing_resource_during_fetch_is_treated_as_remote_delete(self) -> None: - session = self.Session() + session = self.session() tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant") session.add(tenant) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) diff --git a/tests/test_caldav_security.py b/tests/test_caldav_security.py new file mode 100644 index 0000000..993c545 --- /dev/null +++ b/tests/test_caldav_security.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import contextlib +import threading +import unittest +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +from govoplan_calendar.backend.caldav import ( + CalDAVClient, + CalDAVError, + absolute_dav_url, + urllib_transport, +) + + +@contextlib.contextmanager +def running_http_server(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]: + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + yield f"http://{host}:{port}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +class CalDAVUrlSecurityTests(unittest.TestCase): + def test_discovery_href_must_remain_on_configured_origin(self) -> None: + base_url = "https://dav.example.test/calendars/ada/" + + self.assertEqual( + absolute_dav_url(base_url, "/principals/users/ada/"), + "https://dav.example.test/principals/users/ada/", + ) + with self.assertRaisesRegex(CalDAVError, "configured collection origin"): + absolute_dav_url(base_url, "https://evil.example.test/steal/") + with self.assertRaisesRegex(CalDAVError, "query or fragment"): + absolute_dav_url(base_url, "/principals/users/ada/?token=secret") + + def test_object_href_rejects_userinfo_query_and_fragment(self) -> None: + client = CalDAVClient(collection_url="https://dav.example.test/calendars/ada") + + with self.assertRaisesRegex(CalDAVError, "embedded credentials"): + client.object_url("https://user:secret@dav.example.test/calendars/ada/event.ics") + with self.assertRaisesRegex(CalDAVError, "query or fragment"): + client.object_url("/calendars/ada/event.ics?download=1") + with self.assertRaisesRegex(CalDAVError, "query or fragment"): + client.object_url("/calendars/ada/event.ics#fragment") + self.assertEqual( + client.object_url("/calendars/ada/event.ics"), + "https://dav.example.test/calendars/ada/event.ics", + ) + + def test_transport_refuses_redirect_before_forwarding_authorization(self) -> None: + forwarded_authorization: list[str | None] = [] + + class TargetHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + forwarded_authorization.append(self.headers.get("Authorization")) + self.send_response(200) + self.end_headers() + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(TargetHandler) as target_url: + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + self.send_response(302) + self.send_header("Location", f"{target_url}/stolen.ics") + self.end_headers() + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(RedirectHandler) as redirect_url: + status, _headers, _body = urllib_transport( + "GET", + f"{redirect_url}/event.ics", + {"Authorization": "Bearer top-secret"}, + None, + 2, + ) + + self.assertEqual(status, 302) + self.assertEqual(forwarded_authorization, []) + + def test_transport_preserves_same_origin_redirects(self) -> None: + forwarded_authorization: list[str | None] = [] + + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + if self.path == "/event.ics": + self.send_response(302) + self.send_header("Location", "/redirected.ics") + self.end_headers() + return + forwarded_authorization.append(self.headers.get("Authorization")) + self.send_response(200) + self.end_headers() + self.wfile.write(b"calendar") + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(RedirectHandler) as source_url: + status, _headers, body = urllib_transport( + "GET", + f"{source_url}/event.ics", + {"Authorization": "Bearer expected"}, + None, + 2, + ) + + self.assertEqual(status, 200) + self.assertEqual(body, b"calendar") + self.assertEqual(forwarded_authorization, ["Bearer expected"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_http_security.py b/tests/test_http_security.py new file mode 100644 index 0000000..995cf09 --- /dev/null +++ b/tests/test_http_security.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import contextlib +import threading +import unittest +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +from govoplan_calendar.backend.service import CalendarError, http_request + + +@contextlib.contextmanager +def running_http_server(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]: + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + yield f"http://{host}:{port}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +class CalendarHttpSecurityTests(unittest.TestCase): + def test_cross_origin_redirect_does_not_forward_authorization(self) -> None: + forwarded_authorization: list[str | None] = [] + + class TargetHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + forwarded_authorization.append(self.headers.get("Authorization")) + self.send_response(200) + self.end_headers() + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(TargetHandler) as target_url: + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + self.send_response(302) + self.send_header("Location", f"{target_url}/stolen") + self.end_headers() + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(RedirectHandler) as source_url: + with self.assertRaisesRegex(CalendarError, "HTTP 302"): + http_request( + f"{source_url}/feed", + headers={"Authorization": "Bearer top-secret"}, + timeout=2, + ) + + self.assertEqual(forwarded_authorization, []) + + def test_same_origin_redirect_remains_supported(self) -> None: + forwarded_authorization: list[str | None] = [] + + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + if self.path == "/feed": + self.send_response(302) + self.send_header("Location", "/calendar.ics") + self.end_headers() + return + forwarded_authorization.append(self.headers.get("Authorization")) + self.send_response(200) + self.end_headers() + self.wfile.write(b"calendar") + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(RedirectHandler) as source_url: + status, _headers, body = http_request( + f"{source_url}/feed", + headers={"Authorization": "Bearer expected"}, + timeout=2, + ) + + self.assertEqual(status, 200) + self.assertEqual(body, "calendar") + self.assertEqual(forwarded_authorization, ["Bearer expected"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sync_sources.py b/tests/test_sync_sources.py index 92de81a..fc37758 100644 --- a/tests/test_sync_sources.py +++ b/tests/test_sync_sources.py @@ -23,12 +23,17 @@ class CalendarSyncSourceTests(unittest.TestCase): create_scope_tables(self.engine) Base.metadata.create_all(bind=self.engine) self.Session = sessionmaker(bind=self.engine) + self.sessions = [] def tearDown(self) -> None: + for session in reversed(self.sessions): + session.close() Base.metadata.drop_all(bind=self.engine) + self.engine.dispose() def session_with_calendar(self): session = self.Session() + self.sessions.append(session) session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote")) session.flush() @@ -166,6 +171,74 @@ END:VCALENDAR self.assertEqual(event.summary, "Graph item") self.assertIn("Bearer graph-token", request.call_args.kwargs["headers"]["Authorization"]) + def test_graph_sync_rejects_cross_origin_continuation_url(self) -> None: + session, calendar = self.session_with_calendar() + source = create_sync_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarSyncSourceCreateRequest( + source_kind="graph", + calendar_id=calendar.id, + collection_url="me/calendar", + auth_type="bearer", + bearer_token="graph-token", + ), + ) + payload = { + "value": [], + "@odata.nextLink": "https://attacker.example.test/collect?token=secret", + } + + with patch( + "govoplan_calendar.backend.service.http_request", + return_value=(200, {}, json.dumps(payload)), + ) as request: + with self.assertRaisesRegex(CalendarError, "configured source origin"): + sync_source( + session, + tenant_id="tenant-1", + user_id=None, + source_id=source.id, + ) + + self.assertEqual(request.call_count, 1) + self.assertEqual(source.last_status, "error") + + def test_graph_sync_rejects_cross_origin_delta_url(self) -> None: + session, calendar = self.session_with_calendar() + source = create_sync_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarSyncSourceCreateRequest( + source_kind="graph", + calendar_id=calendar.id, + collection_url="me/calendar", + auth_type="bearer", + bearer_token="graph-token", + ), + ) + payload = { + "value": [], + "@odata.deltaLink": "https://attacker.example.test/collect?token=secret", + } + + with patch( + "govoplan_calendar.backend.service.http_request", + return_value=(200, {}, json.dumps(payload)), + ) as request: + with self.assertRaisesRegex(CalendarError, "configured source origin"): + sync_source( + session, + tenant_id="tenant-1", + user_id=None, + source_id=source.id, + ) + + self.assertEqual(request.call_count, 1) + self.assertEqual(source.last_status, "error") + def test_ews_sync_imports_calendar_view_items(self) -> None: session, calendar = self.session_with_calendar() source = create_sync_source(