From c071235ad7565f70dc45ab9e6b2624ecbdecfbe1 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sat, 11 Jul 2026 18:37:51 +0200 Subject: [PATCH] Harden CalDAV sync handling --- src/govoplan_calendar/backend/caldav.py | 15 +++++++++++++-- src/govoplan_calendar/backend/service.py | 5 ++++- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/src/govoplan_calendar/backend/caldav.py b/src/govoplan_calendar/backend/caldav.py index 56f6958..47e6362 100644 --- a/src/govoplan_calendar/backend/caldav.py +++ b/src/govoplan_calendar/backend/caldav.py @@ -310,9 +310,10 @@ 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: + with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated CalDAV HTTP(S) URL. # 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() @@ -422,7 +423,17 @@ def ensure_collection_url(value: str) -> str: value = value.strip() if value and "://" not in value and not value.startswith("/"): value = f"https://{value}" - return value if value.endswith("/") else f"{value}/" + url = validate_http_url(value) + return url if url.endswith("/") else f"{url}/" + + +def validate_http_url(value: str) -> str: + parsed = urllib.parse.urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + 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") + return urllib.parse.urlunparse(parsed) def absolute_dav_url(base_url: str, href: str) -> str: diff --git a/src/govoplan_calendar/backend/service.py b/src/govoplan_calendar/backend/service.py index 38ac6d9..107f785 100644 --- a/src/govoplan_calendar/backend/service.py +++ b/src/govoplan_calendar/backend/service.py @@ -151,6 +151,8 @@ 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: 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") return urllib.parse.urlunparse(parsed) @@ -748,9 +750,10 @@ def http_request( timeout: int = 30, ) -> 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 {}) try: - with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - URLs are admin-configured sync sources. + with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated admin-configured sync source URL. # 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