music-assistant-server

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