633 lines
23 KiB
Python
633 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import posixpath
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Mapping, Protocol
|
|
|
|
from defusedxml import ElementTree as SafeElementTree
|
|
from govoplan_core.security.outbound_http import (
|
|
OutboundHttpError,
|
|
bounded_response_bytes,
|
|
build_outbound_http_opener,
|
|
validate_outbound_http_url,
|
|
)
|
|
|
|
|
|
class CalDAVError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class CalDAVSyncUnsupported(CalDAVError):
|
|
pass
|
|
|
|
|
|
class CalDAVPreconditionFailed(CalDAVError):
|
|
pass
|
|
|
|
|
|
class CalDAVNotFound(CalDAVError):
|
|
pass
|
|
|
|
|
|
class CalDAVTransport(Protocol):
|
|
def __call__(self, method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: int) -> tuple[int, Mapping[str, str], bytes]:
|
|
...
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CalDAVObject:
|
|
href: str
|
|
etag: str | None = None
|
|
calendar_data: str | None = None
|
|
deleted: bool = False
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CalDAVReportResult:
|
|
objects: list[CalDAVObject] = field(default_factory=list)
|
|
sync_token: str | None = None
|
|
ctag: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CalDAVWriteResult:
|
|
href: str
|
|
etag: str | None = None
|
|
status: int = 0
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CalDAVDiscoveryCalendar:
|
|
collection_url: str
|
|
href: str
|
|
display_name: str | None = None
|
|
color: str | None = None
|
|
ctag: str | None = None
|
|
sync_token: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _DAVDiscoveryResponse:
|
|
href: str
|
|
display_name: str | None = None
|
|
color: str | None = None
|
|
ctag: str | None = None
|
|
sync_token: str | None = None
|
|
is_calendar: bool = False
|
|
principal_hrefs: tuple[str, ...] = ()
|
|
calendar_home_set_hrefs: tuple[str, ...] = ()
|
|
supported_components: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class _DAVDiscoveryDraft:
|
|
display_name: str | None = None
|
|
color: str | None = None
|
|
ctag: str | None = None
|
|
sync_token: str | None = None
|
|
is_calendar: bool = False
|
|
principal_hrefs: list[str] = field(default_factory=list)
|
|
calendar_home_set_hrefs: list[str] = field(default_factory=list)
|
|
supported_components: list[str] = field(default_factory=list)
|
|
|
|
|
|
class CalDAVClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
collection_url: str,
|
|
username: str | None = None,
|
|
password: str | None = None,
|
|
bearer_token: str | None = None,
|
|
timeout: int = 30,
|
|
transport: CalDAVTransport | None = None,
|
|
) -> None:
|
|
self.collection_url = ensure_collection_url(collection_url)
|
|
self.username = username
|
|
self.password = password
|
|
self.bearer_token = bearer_token
|
|
self.timeout = timeout
|
|
self.transport = transport or urllib_transport
|
|
|
|
def propfind_collection(self) -> CalDAVReportResult:
|
|
body = b"""<?xml version="1.0" encoding="utf-8"?>
|
|
<D:propfind xmlns:D="DAV:" xmlns:CS="http://calendarserver.org/ns/">
|
|
<D:prop>
|
|
<D:displayname/>
|
|
<D:sync-token/>
|
|
<CS:getctag/>
|
|
</D:prop>
|
|
</D:propfind>"""
|
|
payload = self.request("PROPFIND", self.collection_url, body=body, depth="0", expected={207})
|
|
return parse_multistatus(payload)
|
|
|
|
def discover_calendars(self) -> list[CalDAVDiscoveryCalendar]:
|
|
start_url = self.collection_url
|
|
calendars: dict[str, CalDAVDiscoveryCalendar] = {}
|
|
home_urls: list[str] = []
|
|
principal_urls: list[str] = []
|
|
visited_urls: set[tuple[str, str]] = set()
|
|
errors: list[str] = []
|
|
|
|
def add_home_href(base_url: str, href: str) -> None:
|
|
url = ensure_collection_url(absolute_dav_url(base_url, href))
|
|
if url not in home_urls:
|
|
home_urls.append(url)
|
|
|
|
def add_principal_href(base_url: str, href: str) -> None:
|
|
url = ensure_collection_url(absolute_dav_url(base_url, href))
|
|
if url not in principal_urls:
|
|
principal_urls.append(url)
|
|
|
|
def add_calendar(base_url: str, response: _DAVDiscoveryResponse) -> None:
|
|
if not response.is_calendar:
|
|
return
|
|
if response.supported_components and "VEVENT" not in response.supported_components:
|
|
return
|
|
url = ensure_collection_url(absolute_dav_url(base_url, response.href or base_url))
|
|
calendars[url] = CalDAVDiscoveryCalendar(
|
|
collection_url=url,
|
|
href=response.href,
|
|
display_name=response.display_name,
|
|
color=response.color,
|
|
ctag=response.ctag,
|
|
sync_token=response.sync_token,
|
|
)
|
|
|
|
def propfind(url: str, depth: str) -> list[_DAVDiscoveryResponse]:
|
|
key = (url, depth)
|
|
if key in visited_urls:
|
|
return []
|
|
visited_urls.add(key)
|
|
return self.propfind_discovery(url, depth=depth)
|
|
|
|
try:
|
|
for response in propfind(start_url, "0"):
|
|
add_calendar(start_url, response)
|
|
for href in response.calendar_home_set_hrefs:
|
|
add_home_href(start_url, href)
|
|
for href in response.principal_hrefs:
|
|
add_principal_href(start_url, href)
|
|
except CalDAVError as exc:
|
|
errors.append(str(exc))
|
|
|
|
for principal_url in principal_urls[:6]:
|
|
try:
|
|
for response in propfind(principal_url, "0"):
|
|
for href in response.calendar_home_set_hrefs:
|
|
add_home_href(principal_url, href)
|
|
except CalDAVError as exc:
|
|
errors.append(str(exc))
|
|
|
|
if not home_urls:
|
|
home_urls.append(start_url)
|
|
|
|
for home_url in home_urls:
|
|
try:
|
|
for response in propfind(home_url, "1"):
|
|
add_calendar(home_url, response)
|
|
except CalDAVError as exc:
|
|
errors.append(str(exc))
|
|
|
|
if not calendars and errors:
|
|
raise CalDAVError(f"CalDAV discovery did not find any calendar collections: {errors[0]}")
|
|
return sorted(calendars.values(), key=lambda item: ((item.display_name or item.collection_url).lower(), item.collection_url))
|
|
|
|
def propfind_discovery(self, url: str, *, depth: str) -> list[_DAVDiscoveryResponse]:
|
|
body = b"""<?xml version="1.0" encoding="utf-8"?>
|
|
<D:propfind xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav" xmlns:CS="http://calendarserver.org/ns/" xmlns:ICAL="http://apple.com/ns/ical/">
|
|
<D:prop>
|
|
<D:displayname/>
|
|
<D:current-user-principal/>
|
|
<D:principal-URL/>
|
|
<D:resourcetype/>
|
|
<D:sync-token/>
|
|
<C:calendar-home-set/>
|
|
<C:supported-calendar-component-set/>
|
|
<CS:getctag/>
|
|
<ICAL:calendar-color/>
|
|
</D:prop>
|
|
</D:propfind>"""
|
|
payload = self.request("PROPFIND", ensure_collection_url(url), body=body, depth=depth, expected={207})
|
|
return parse_discovery_multistatus(payload)
|
|
|
|
def list_objects(self) -> CalDAVReportResult:
|
|
body = b"""<?xml version="1.0" encoding="utf-8"?>
|
|
<C:calendar-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
|
|
<D:prop>
|
|
<D:getetag/>
|
|
<C:calendar-data/>
|
|
</D:prop>
|
|
<C:filter>
|
|
<C:comp-filter name="VCALENDAR">
|
|
<C:comp-filter name="VEVENT"/>
|
|
</C:comp-filter>
|
|
</C:filter>
|
|
</C:calendar-query>"""
|
|
payload = self.request("REPORT", self.collection_url, body=body, depth="1", expected={207})
|
|
return parse_multistatus(payload)
|
|
|
|
def sync_collection(self, sync_token: str) -> CalDAVReportResult:
|
|
body = f"""<?xml version="1.0" encoding="utf-8"?>
|
|
<D:sync-collection xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
|
|
<D:sync-token>{xml_escape(sync_token)}</D:sync-token>
|
|
<D:sync-level>1</D:sync-level>
|
|
<D:prop>
|
|
<D:getetag/>
|
|
<C:calendar-data/>
|
|
</D:prop>
|
|
</D:sync-collection>""".encode("utf-8")
|
|
try:
|
|
payload = self.request("REPORT", self.collection_url, body=body, depth="1", expected={207})
|
|
except CalDAVError as exc:
|
|
raise CalDAVSyncUnsupported(str(exc)) from exc
|
|
return parse_multistatus(payload)
|
|
|
|
def fetch_object(self, href: str) -> str:
|
|
return self.fetch_object_state(href).calendar_data or ""
|
|
|
|
def fetch_object_state(self, href: str) -> CalDAVObject:
|
|
"""Fetch a resource together with its ETag for outbox reconciliation."""
|
|
|
|
_status, headers, payload = self.request_raw(
|
|
"GET",
|
|
self.object_url(href),
|
|
body=None,
|
|
depth=None,
|
|
expected={200},
|
|
)
|
|
return CalDAVObject(
|
|
href=href,
|
|
etag=response_etag(headers),
|
|
calendar_data=payload.decode("utf-8"),
|
|
)
|
|
|
|
def put_object(self, href: str, ics: str, *, etag: str | None = None, create: bool = False, overwrite: bool = False) -> CalDAVWriteResult:
|
|
headers = {"Content-Type": "text/calendar; charset=utf-8"}
|
|
if create:
|
|
headers["If-None-Match"] = "*"
|
|
elif etag and not overwrite:
|
|
headers["If-Match"] = etag
|
|
elif not overwrite:
|
|
raise CalDAVPreconditionFailed("Refusing CalDAV PUT without an ETag")
|
|
status, response_headers, _payload = self.request_raw(
|
|
"PUT",
|
|
self.object_url(href),
|
|
body=ics.encode("utf-8"),
|
|
depth=None,
|
|
expected={200, 201, 204},
|
|
extra_headers=headers,
|
|
)
|
|
return CalDAVWriteResult(href=href, etag=response_etag(response_headers), status=status)
|
|
|
|
def delete_object(self, href: str, *, etag: str | None = None, overwrite: bool = False) -> CalDAVWriteResult:
|
|
headers: dict[str, str] = {}
|
|
if etag and not overwrite:
|
|
headers["If-Match"] = etag
|
|
elif not overwrite:
|
|
raise CalDAVPreconditionFailed("Refusing CalDAV DELETE without an ETag")
|
|
status, response_headers, _payload = self.request_raw(
|
|
"DELETE",
|
|
self.object_url(href),
|
|
body=None,
|
|
depth=None,
|
|
expected={200, 202, 204, 404},
|
|
extra_headers=headers,
|
|
)
|
|
return CalDAVWriteResult(href=href, etag=response_etag(response_headers), status=status)
|
|
|
|
def object_url(self, href: str) -> str:
|
|
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)
|
|
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("/") + "/"
|
|
if not candidate_path.startswith(collection_prefix) or candidate_path == collection_path:
|
|
raise CalDAVError("CalDAV object href must remain inside the configured collection path")
|
|
if candidate_parts.query or candidate_parts.fragment:
|
|
raise CalDAVError("CalDAV object href must not contain a query or fragment")
|
|
return urllib.parse.urlunparse(candidate_parts)
|
|
|
|
def request(self, method: str, url: str, *, body: bytes | None, depth: str | None, expected: set[int]) -> bytes:
|
|
_status, _headers, payload = self.request_raw(method, url, body=body, depth=depth, expected=expected)
|
|
return payload
|
|
|
|
def request_raw(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
body: bytes | None,
|
|
depth: str | None,
|
|
expected: set[int],
|
|
extra_headers: Mapping[str, str] | None = None,
|
|
) -> tuple[int, Mapping[str, str], bytes]:
|
|
headers: dict[str, str] = {
|
|
"Accept": "application/xml,text/calendar,*/*",
|
|
"User-Agent": "govoplan-calendar-caldav/0.1",
|
|
}
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/xml; charset=utf-8"
|
|
if depth is not None:
|
|
headers["Depth"] = depth
|
|
if self.bearer_token:
|
|
headers["Authorization"] = f"Bearer {self.bearer_token}"
|
|
elif self.username and self.password:
|
|
token = base64.b64encode(f"{self.username}:{self.password}".encode("utf-8")).decode("ascii")
|
|
headers["Authorization"] = f"Basic {token}"
|
|
if extra_headers:
|
|
headers.update(dict(extra_headers))
|
|
status, response_headers, payload = self.transport(method, url, headers, body, self.timeout)
|
|
if status not in expected:
|
|
if status == 404:
|
|
raise CalDAVNotFound(f"{method} {url} returned HTTP {status}")
|
|
if status == 412:
|
|
raise CalDAVPreconditionFailed(f"{method} {url} failed because the remote resource changed")
|
|
raise CalDAVError(f"{method} {url} returned HTTP {status}")
|
|
return status, response_headers, payload
|
|
|
|
|
|
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:
|
|
url = validate_outbound_http_url(url, label="CalDAV URL")
|
|
request = urllib.request.Request( # noqa: S310 - URL is validated and origin-confined.
|
|
url,
|
|
data=body,
|
|
headers=dict(headers),
|
|
method=method,
|
|
)
|
|
opener = build_outbound_http_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
|
|
response_headers = dict(response.headers.items())
|
|
return response.status, response_headers, bounded_response_bytes(
|
|
response,
|
|
headers=response_headers,
|
|
label="CalDAV response",
|
|
)
|
|
except urllib.error.HTTPError as exc:
|
|
response_headers = dict(exc.headers.items())
|
|
try:
|
|
payload = bounded_response_bytes(exc, headers=response_headers, label="CalDAV error response")
|
|
except OutboundHttpError as policy_exc:
|
|
raise CalDAVError(f"{method} {url} failed: {policy_exc}") from policy_exc
|
|
return exc.code, response_headers, payload
|
|
except urllib.error.URLError as exc:
|
|
raise CalDAVError(f"{method} {url} failed: {exc.reason}") from exc
|
|
except (OutboundHttpError, ValueError) as exc:
|
|
raise CalDAVError(f"{method} {url} failed: {exc}") from exc
|
|
|
|
|
|
def parse_multistatus(payload: bytes) -> CalDAVReportResult:
|
|
try:
|
|
root = SafeElementTree.fromstring(payload)
|
|
except SafeElementTree.ParseError as exc:
|
|
raise CalDAVError(f"Invalid CalDAV XML response: {exc}") from exc
|
|
objects: list[CalDAVObject] = []
|
|
sync_token = first_child_text(root, "sync-token")
|
|
ctag = first_child_text(root, "getctag")
|
|
for response in child_elements(root, "response"):
|
|
href = first_child_text(response, "href")
|
|
if not href:
|
|
continue
|
|
deleted = False
|
|
etag = None
|
|
calendar_data = None
|
|
for propstat in child_elements(response, "propstat"):
|
|
status = first_child_text(propstat, "status") or ""
|
|
prop = first_child(propstat, "prop")
|
|
if prop is None:
|
|
continue
|
|
if " 404 " in status or status.endswith(" 404"):
|
|
deleted = True
|
|
continue
|
|
if " 200 " not in status and not status.endswith(" 200"):
|
|
continue
|
|
etag = first_child_text(prop, "getetag") or etag
|
|
calendar_data = first_child_text(prop, "calendar-data") or calendar_data
|
|
sync_token = first_child_text(prop, "sync-token") or sync_token
|
|
ctag = first_child_text(prop, "getctag") or ctag
|
|
objects.append(CalDAVObject(href=href, etag=strip_weak_etag(etag), calendar_data=calendar_data, deleted=deleted))
|
|
return CalDAVReportResult(objects=objects, sync_token=sync_token, ctag=ctag)
|
|
|
|
|
|
def parse_discovery_multistatus(payload: bytes) -> list[_DAVDiscoveryResponse]:
|
|
try:
|
|
root = SafeElementTree.fromstring(payload)
|
|
except SafeElementTree.ParseError as exc:
|
|
raise CalDAVError(f"Invalid CalDAV XML response: {exc}") from exc
|
|
return [
|
|
parsed
|
|
for response in child_elements(root, "response")
|
|
if (parsed := _parse_discovery_response(response)) is not None
|
|
]
|
|
|
|
|
|
def _parse_discovery_response(response: Any) -> _DAVDiscoveryResponse | None:
|
|
href = first_child_text(response, "href")
|
|
if not href:
|
|
return None
|
|
draft = _DAVDiscoveryDraft()
|
|
for propstat in child_elements(response, "propstat"):
|
|
_apply_discovery_propstat(draft, propstat)
|
|
return _DAVDiscoveryResponse(
|
|
href=href,
|
|
display_name=draft.display_name,
|
|
color=draft.color,
|
|
ctag=draft.ctag,
|
|
sync_token=draft.sync_token,
|
|
is_calendar=draft.is_calendar,
|
|
principal_hrefs=dedupe_tuple(draft.principal_hrefs),
|
|
calendar_home_set_hrefs=dedupe_tuple(draft.calendar_home_set_hrefs),
|
|
supported_components=dedupe_tuple(draft.supported_components),
|
|
)
|
|
|
|
|
|
def _apply_discovery_propstat(draft: _DAVDiscoveryDraft, propstat: Any) -> None:
|
|
if not discovery_propstat_is_success(propstat):
|
|
return
|
|
prop = first_child(propstat, "prop")
|
|
if prop is None:
|
|
return
|
|
for item in prop:
|
|
_apply_discovery_property(draft, item)
|
|
|
|
|
|
def discovery_propstat_is_success(propstat: Any) -> bool:
|
|
status = first_child_text(propstat, "status") or ""
|
|
return not status or " 200 " in status or status.endswith(" 200") or " 207 " in status
|
|
|
|
|
|
def _apply_discovery_property(draft: _DAVDiscoveryDraft, item: Any) -> None:
|
|
name = local_name(item.tag)
|
|
text = item.text.strip() if item.text else ""
|
|
if name == "displayname" and text:
|
|
draft.display_name = text or draft.display_name
|
|
elif name == "calendar-color" and text:
|
|
draft.color = text or draft.color
|
|
elif name == "getctag" and text:
|
|
draft.ctag = text or draft.ctag
|
|
elif name == "sync-token" and text:
|
|
draft.sync_token = text or draft.sync_token
|
|
elif name == "resourcetype":
|
|
draft.is_calendar = draft.is_calendar or discovery_resource_is_calendar(item)
|
|
elif name in {"current-user-principal", "principal-URL"}:
|
|
draft.principal_hrefs.extend(nested_href_texts(item))
|
|
elif name == "calendar-home-set":
|
|
draft.calendar_home_set_hrefs.extend(nested_href_texts(item))
|
|
elif name == "supported-calendar-component-set":
|
|
draft.supported_components.extend(supported_calendar_component_names(item))
|
|
|
|
|
|
def discovery_resource_is_calendar(item: Any) -> bool:
|
|
return any(local_name(child.tag) == "calendar" for child in item)
|
|
|
|
|
|
def supported_calendar_component_names(item: Any) -> list[str]:
|
|
names: list[str] = []
|
|
for component in item:
|
|
if local_name(component.tag) != "comp":
|
|
continue
|
|
component_name = (component.attrib.get("name") or "").strip().upper()
|
|
if component_name:
|
|
names.append(component_name)
|
|
return names
|
|
|
|
|
|
def dedupe_tuple(values: list[str]) -> tuple[str, ...]:
|
|
return tuple(dict.fromkeys(values))
|
|
|
|
|
|
def ensure_collection_url(value: str) -> str:
|
|
value = value.strip()
|
|
if value and "://" not in value and not value.startswith("/"):
|
|
value = f"https://{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.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 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)
|
|
candidate = validate_outbound_http_url(candidate, label="CalDAV redirect URL")
|
|
except (CalDAVError, OutboundHttpError):
|
|
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:
|
|
if value is None:
|
|
return None
|
|
return value.strip()
|
|
|
|
|
|
def response_etag(headers: Mapping[str, str]) -> str | None:
|
|
for key, value in headers.items():
|
|
if key.lower() == "etag" and value:
|
|
return strip_weak_etag(value)
|
|
return None
|
|
|
|
|
|
def xml_escape(value: str) -> str:
|
|
return value.replace("&", "&").replace("<", "<").replace(">", ">")
|
|
|
|
|
|
def first_child(element: Any, name: str) -> Any | None:
|
|
for child in element:
|
|
if local_name(child.tag) == name:
|
|
return child
|
|
return None
|
|
|
|
|
|
def first_child_text(element: Any, name: str) -> str | None:
|
|
found = first_child(element, name)
|
|
if found is None or found.text is None:
|
|
return None
|
|
return found.text.strip()
|
|
|
|
|
|
def child_elements(element: Any, name: str) -> list[Any]:
|
|
return [child for child in element if local_name(child.tag) == name]
|
|
|
|
|
|
def nested_href_texts(element: Any) -> list[str]:
|
|
hrefs: list[str] = []
|
|
for child in element.iter():
|
|
if local_name(child.tag) == "href" and child.text and child.text.strip():
|
|
hrefs.append(child.text.strip())
|
|
return hrefs
|
|
|
|
|
|
def local_name(tag: str) -> str:
|
|
return tag.rsplit("}", 1)[-1] if "}" in tag else tag
|