Harden CalDAV sync handling
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user