/
/
/
1"""WebDAV helper functions for Music Assistant."""
2
3from __future__ import annotations
4
5import contextlib
6import logging
7from dataclasses import dataclass
8from urllib.parse import quote, unquote, urljoin
9
10import aiohttp
11from defusedxml import ElementTree
12from music_assistant_models.errors import LoginFailed, ProviderUnavailableError, SetupFailedError
13
14LOGGER = logging.getLogger(__name__)
15
16DAV_NAMESPACE = {"d": "DAV:"}
17
18PROPFIND_BODY = """<?xml version="1.0" encoding="utf-8"?>
19<d:propfind xmlns:d="DAV:">
20 <d:prop>
21 <d:resourcetype/>
22 <d:getcontentlength/>
23 <d:getlastmodified/>
24 <d:displayname/>
25 <d:getetag/>
26 </d:prop>
27</d:propfind>"""
28
29
30@dataclass
31class WebDAVItem:
32 """Representation of a WebDAV resource."""
33
34 href: str
35 name: str
36 is_dir: bool
37 size: int | None = None
38 last_modified: str | None = None
39 etag: str | None = None
40
41
42async def webdav_propfind(
43 session: aiohttp.ClientSession,
44 url: str,
45 depth: int = 1,
46 timeout: int = 30,
47 auth_header: str | None = None,
48) -> list[WebDAVItem]:
49 """
50 Execute a PROPFIND request on a WebDAV resource.
51
52 :param session: Active HTTP session.
53 :param url: WebDAV URL to query.
54 :param depth: Depth level (0=properties only, 1=immediate children).
55 :param timeout: Request timeout in seconds.
56 :param auth_header: Optional pre-encoded Authorization header value (e.g. "Basic ...").
57 :returns: List of WebDAVItem objects.
58 :raises LoginFailed: Authentication failed (401/403).
59 :raises SetupFailedError: Server error during setup.
60 :raises ProviderUnavailableError: Connection or timeout error.
61 """
62 headers = {"Depth": str(depth), "Content-Type": "application/xml; charset=utf-8"}
63 if auth_header:
64 headers["Authorization"] = auth_header
65
66 try:
67 async with session.request(
68 "PROPFIND",
69 url,
70 headers=headers,
71 data=PROPFIND_BODY,
72 timeout=aiohttp.ClientTimeout(total=timeout),
73 ) as resp:
74 if resp.status == 401:
75 raise LoginFailed("Authentication failed for WebDAV server")
76 if resp.status == 403:
77 raise LoginFailed("Access forbidden for WebDAV server")
78 if resp.status == 404:
79 return []
80 if resp.status >= 400:
81 raise SetupFailedError(f"WebDAV PROPFIND failed with status {resp.status}")
82
83 response_text = await resp.text()
84 return _parse_propfind_response(response_text, url)
85
86 except TimeoutError as err:
87 raise ProviderUnavailableError(
88 f"WebDAV connection timeout: {url}",
89 translation_key="connection_timeout",
90 translation_args=[url],
91 ) from err
92 except aiohttp.ClientError as err:
93 raise ProviderUnavailableError(f"WebDAV connection error: {err}") from err
94
95
96def _find_prop(props: list[ElementTree.Element], tag: str) -> ElementTree.Element | None:
97 """Return the first match for tag across a response's merged propstat prop elements."""
98 for prop in props:
99 if (elem := prop.find(tag, DAV_NAMESPACE)) is not None:
100 return elem
101 return None
102
103
104def _parse_propfind_response(response_text: str, base_url: str) -> list[WebDAVItem]:
105 """Parse WebDAV PROPFIND XML response."""
106 try:
107 root = ElementTree.fromstring(response_text)
108 except ElementTree.ParseError as err:
109 LOGGER.warning("Failed to parse WebDAV PROPFIND response: %s", err)
110 return []
111
112 items: list[WebDAVItem] = []
113 base_url_normalized = base_url.rstrip("/")
114
115 for response_elem in root.findall("d:response", DAV_NAMESPACE):
116 href_elem = response_elem.find("d:href", DAV_NAMESPACE)
117 if href_elem is None or not href_elem.text:
118 continue
119
120 href = unquote(href_elem.text.rstrip("/"))
121
122 # Skip the base directory itself
123 if href.rstrip("/") == base_url_normalized:
124 continue
125
126 # a server may split properties it cannot satisfy (e.g. an unsupported getetag) into
127 # a separate propstat with a non-2xx status; merge every successful block's props so a
128 # 404 block returned first does not shadow resourcetype/getlastmodified from a later 200
129 props: list[ElementTree.Element] = []
130 for propstat in response_elem.findall("d:propstat", DAV_NAMESPACE):
131 status_elem = propstat.find("d:status", DAV_NAMESPACE)
132 if status_elem is not None and status_elem.text and " 200 " not in status_elem.text:
133 continue
134 if (prop := propstat.find("d:prop", DAV_NAMESPACE)) is not None:
135 props.append(prop)
136 if not props:
137 continue
138
139 # Check if it's a directory
140 resourcetype = _find_prop(props, "d:resourcetype")
141 is_collection = (
142 resourcetype is not None
143 and resourcetype.find("d:collection", DAV_NAMESPACE) is not None
144 )
145
146 # Get size (only for files)
147 size = None
148 if not is_collection:
149 contentlength = _find_prop(props, "d:getcontentlength")
150 if contentlength is not None and contentlength.text:
151 with contextlib.suppress(ValueError):
152 size = int(contentlength.text)
153
154 # Get last modified
155 lastmodified = _find_prop(props, "d:getlastmodified")
156 last_modified = lastmodified.text if lastmodified is not None else None
157
158 # Get etag (used only as a higher-precision metadata-file change token)
159 etagelem = _find_prop(props, "d:getetag")
160 etag = None
161 if etagelem is not None and etagelem.text:
162 etag = etagelem.text.strip().removeprefix("W/").strip('"') or None
163
164 # Get display name or extract from href
165 displayname = _find_prop(props, "d:displayname")
166 if displayname is not None and displayname.text:
167 name = displayname.text
168 else:
169 name = href.split("/")[-1] or href.split("/")[-2]
170
171 items.append(
172 WebDAVItem(
173 href=href,
174 name=name,
175 is_dir=is_collection,
176 size=size,
177 last_modified=last_modified,
178 etag=etag,
179 )
180 )
181
182 return items
183
184
185async def webdav_test_connection(
186 session: aiohttp.ClientSession,
187 base_url: str,
188 username: str | None,
189 password: str | None,
190 timeout: int = 10,
191) -> None:
192 """
193 Test WebDAV connection and authentication.
194
195 :param session: Active HTTP session.
196 :param base_url: WebDAV server URL.
197 :param username: Optional username.
198 :param password: Optional password.
199 :param timeout: Connection timeout in seconds.
200 :raises LoginFailed: Authentication failed.
201 :raises SetupFailedError: Connection or configuration error.
202 """
203 auth_header = aiohttp.encode_basic_auth(username, password or "") if username else None
204
205 try:
206 await webdav_propfind(session, base_url, depth=0, timeout=timeout, auth_header=auth_header)
207 except ProviderUnavailableError as err:
208 # During setup, connection errors should be SetupFailedError
209 raise SetupFailedError(str(err)) from err
210
211
212def build_webdav_url(base_url: str, path: str) -> str:
213 """
214 Build a WebDAV URL by joining the base URL with a relative resource path.
215
216 :param base_url: The WebDAV base URL.
217 :param path: A relative resource path, or an absolute URL which is returned as-is.
218 """
219 if path.startswith(("http://", "https://")):
220 return path
221 normalized_base = base_url if base_url.endswith("/") else f"{base_url}/"
222 # Percent-encode the path so reserved characters (e.g. ; ? # :) survive intact;
223 # left unencoded they would be misread as URL params/query/fragment/scheme.
224 quoted_path = quote(path.removeprefix("/"), safe="/")
225 return urljoin(normalized_base, quoted_path)
226