Coverage for kataglyphis_webdavclient/webdavclient.py: 95%
201 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 00:43 +0000
1# Copyright (c) 2024 Jonas Heinle
3"""WebDAV client implementation for listing and downloading remote files."""
5import sys
6import threading
7import time
8import urllib.parse
9import uuid
10from pathlib import Path, PurePosixPath
12import requests
13from defusedxml import ElementTree
14from loguru import logger
15from requests.auth import HTTPBasicAuth
18REQUEST_TIMEOUT_SECONDS = 30
19LOG_FILE = "logs/downloadMd_s.log"
21# One loguru sink per log file: a sink per client wrote every line once per client and rotated the file N times.
22_log_sinks: set[str] = set()
23_log_sinks_lock = threading.Lock()
26def _add_log_sink(log_file: str) -> None:
27 path = str(Path(log_file).absolute())
28 with _log_sinks_lock:
29 if path in _log_sinks:
30 return
31 logger.add(path, rotation="500 MB")
32 _log_sinks.add(path)
35def _join_remote_url(*parts: str) -> str:
36 cleaned_parts = [part.strip("/") for part in parts if part]
37 return "/".join(cleaned_parts)
40def _remote_relative_path(href: str, base_url: str) -> tuple[str, ...]:
41 """Return validated, decoded path components below the same-origin DAV base."""
42 if any(ord(char) < 32 for char in href):
43 message = "WebDAV href contains control characters"
44 raise ValueError(message)
45 raw_path = urllib.parse.unquote(urllib.parse.urlsplit(href).path, errors="strict")
46 if any(part in {".", ".."} for part in raw_path.split("/")):
47 message = "WebDAV href contains an unsafe path component"
48 raise ValueError(message)
49 base = urllib.parse.urlsplit(base_url.rstrip("/") + "/")
50 remote = urllib.parse.urlsplit(urllib.parse.urljoin(base.geturl(), href))
51 if (remote.scheme, remote.netloc) != (base.scheme, base.netloc):
52 message = "WebDAV href belongs to another origin"
53 raise ValueError(message)
54 if remote.query or remote.fragment:
55 message = "WebDAV href contains a query or fragment"
56 raise ValueError(message)
57 path = urllib.parse.unquote(remote.path, errors="strict")
58 base_path = urllib.parse.unquote(base.path, errors="strict").rstrip("/")
59 parts = tuple(path.strip("/").split("/")) if path.strip("/") else ()
60 base_parts = tuple(base_path.strip("/").split("/")) if base_path else ()
61 # Validate before prefix matching so neither dot segments nor Windows syntax can change the destination.
62 if any(
63 not part
64 or part in {".", ".."}
65 or any(ord(char) < 32 or char in '\\:<>"|?*' for char in part)
66 or part.endswith((".", " "))
67 or part.split(".", 1)[0].rstrip(" ").upper()
68 in {
69 "CON",
70 "PRN",
71 "AUX",
72 "NUL",
73 *(f"COM{i}" for i in "123456789¹²³"),
74 *(f"LPT{i}" for i in "123456789¹²³"),
75 }
76 for part in parts
77 ):
78 message = "WebDAV href contains an unsafe path component"
79 raise ValueError(message)
80 if parts[: len(base_parts)] != base_parts:
81 message = "WebDAV href is outside the remote base path"
82 raise ValueError(message)
83 return parts[len(base_parts) :]
86def _download_target(base: Path, relative: tuple[str, ...]) -> Path:
87 """Resolve existing symlinks and reject a destination outside the download root."""
88 target = base.joinpath(*relative).resolve()
89 if not relative or not target.is_relative_to(base):
90 message = "WebDAV download target is outside the local base path"
91 raise ValueError(message)
92 return target
95def _replace(part: Path, target: Path) -> None:
96 # Windows refuses a replace while another thread replaces the same target (PermissionError); it clears in milliseconds.
97 attempts = 40 if sys.platform == "win32" else 1
98 for attempt in range(attempts):
99 try:
100 part.replace(target)
101 except PermissionError:
102 if attempt == attempts - 1:
103 raise
104 time.sleep(0.005 * (attempt + 1))
105 else:
106 return
109def _write_atomically(target: Path, response: requests.Response) -> None:
110 # A private file renamed into place: two downloads of one file never interleave, and a reader never sees half of it.
111 part = target.with_name(f".{target.name}.{uuid.uuid4().hex}.part")
112 try:
113 with part.open("xb") as f:
114 f.writelines(response.iter_content(chunk_size=8192))
115 _replace(part, target)
116 except BaseException:
117 part.unlink(missing_ok=True)
118 raise
121class WebDavClient:
122 """A simple WebDav client for listing and downloading files and folders."""
124 def __init__(self, hostname: str, username: str, password: str) -> None:
125 """Initialize a client with credentials and logging output folder."""
126 self.hostname: str = hostname
127 self.username: str = username
128 self.password: str = password
129 self.auth: HTTPBasicAuth = HTTPBasicAuth(
130 username,
131 password,
132 )
134 self.ensure_folder_exists(str(Path(LOG_FILE).parent))
135 _add_log_sink(LOG_FILE)
137 def _hrefs(self, url: str) -> list[str]:
138 headers: dict[str, str] = {"Content-Type": "application/xml", "Depth": "1"}
139 with requests.request(
140 "PROPFIND",
141 url,
142 auth=self.auth,
143 headers=headers,
144 timeout=REQUEST_TIMEOUT_SECONDS,
145 ) as response:
146 if response.status_code != 207:
147 message = f"Failed to list {url}: HTTP {response.status_code}"
148 raise OSError(message)
149 tree = ElementTree.fromstring(response.content)
150 hrefs = []
151 for item in tree.findall("{DAV:}response"):
152 href = item.findtext("{DAV:}href")
153 if not href:
154 message = "WebDAV listing contains a response without an href"
155 raise ValueError(message)
156 hrefs.append(href)
157 return hrefs
159 def list_files(self, url: str) -> list[str]:
160 """List the files directly below a WebDAV URL; OSError unless PROPFIND answers 207."""
161 files = []
162 for href in self._hrefs(url):
163 if not href.endswith("/"):
164 logger.debug("Found file: {} for the following url: {}", href, url)
165 files.append(href)
166 return files
168 def list_folders(self, remote_base_path: str) -> list[str]:
169 """List the non-hidden folders directly below a remote base path; OSError unless PROPFIND answers 207."""
170 url: str = _join_remote_url(self.hostname, remote_base_path)
171 folders = []
172 for href in self._hrefs(url):
173 if not href.endswith("/"):
174 continue
175 relative = _remote_relative_path(href, url)
176 if not relative:
177 continue
178 if len(relative) != 1:
179 message = "WebDAV folder is not a direct child of its parent"
180 raise ValueError(message)
181 if not relative[0].startswith("."):
182 folders.append(urllib.parse.quote(relative[0], safe=""))
183 return folders
185 def filter_after_global_base_path(self, path: str, remote_base_path: str) -> str:
186 """Return what follows /remote_base_path/ in path, or path unchanged when it is absent."""
187 search_str = "/" + remote_base_path + "/"
188 if search_str in path:
189 logger.debug(
190 "Found folder {} for path {} and remote base path: {}",
191 search_str,
192 path,
193 remote_base_path,
194 )
195 url_after_removing_everything_before_and_including_remote_base_name = (
196 path.split(search_str, 1)[1]
197 )
198 logger.info(
199 "Folder structure everything after the remote_base_path is: {}",
200 url_after_removing_everything_before_and_including_remote_base_name,
201 )
202 return url_after_removing_everything_before_and_including_remote_base_name
203 logger.error("Could not find search string: {} in path: {}", search_str, path)
204 return path
206 def ensure_folder_exists(self, path: str) -> None:
207 """Ensure that the given folder exists."""
208 folder_path = Path(path)
209 if not folder_path.exists():
210 folder_path.mkdir(parents=True, exist_ok=True)
211 logger.debug("Folder created: {}", path)
212 else:
213 logger.debug("Folder already exists: {}", path)
215 def get_sub_path(self, full_path: str, initial_part: str) -> str:
216 """Return the part of a host-less full_path after initial_part; ValueError when it is absent."""
217 logger.debug("We are in the 'get_sub_path' method.")
218 decoded_full_path = urllib.parse.unquote(full_path)
219 logger.debug("The decoded full file path is: {}", decoded_full_path)
220 decoded_initial_part = urllib.parse.unquote(initial_part)
221 logger.debug("The decoded initial file path is: {}", decoded_initial_part)
222 # A trailing slash makes the match a whole path segment.
223 if not decoded_initial_part.endswith("/"):
224 decoded_initial_part += "/"
226 start_idx = decoded_full_path.find(decoded_initial_part)
228 if start_idx == -1:
229 logger.error(
230 "The {} string is not in the full_path={}", initial_part, full_path
231 )
232 error_details = "The full path does not contain the initial part."
233 raise ValueError(error_details)
235 # Handle the edge case where the full path is exactly the initial part
236 decoded_initial_part = "/" + decoded_initial_part
237 if full_path == decoded_initial_part.rstrip("/"):
238 logger.debug("The get_sub_path() method returns empty string")
239 return ""
241 if full_path.startswith(initial_part):
242 logger.debug(
243 "The get_sub_path() method returns {}", full_path[len(initial_part) :]
244 )
245 return full_path[len(initial_part) :]
246 sub_path_start_idx = start_idx + len(decoded_initial_part) - 1
248 sub_path = decoded_full_path[sub_path_start_idx:]
250 logger.debug("The get_sub_path() method returns {}", sub_path)
252 return sub_path
254 def download_files(
255 self,
256 global_remote_base_path: str,
257 remote_base_path: str,
258 local_base_path: str,
259 *,
260 extension: str | None = None,
261 ) -> int:
262 """Atomically download matching files, preserving paths below the global base; return the count."""
263 base = Path(local_base_path).resolve()
264 base.mkdir(parents=True, exist_ok=True)
265 global_url = _join_remote_url(self.hostname, global_remote_base_path)
266 url = _join_remote_url(self.hostname, remote_base_path)
267 files_on_host = self.list_files(url)
268 written = 0
269 for file_path in files_on_host:
270 relative = _remote_relative_path(file_path, global_url)
271 local_file_path = _download_target(base, relative)
272 if extension and not relative[-1].lower().endswith(extension.lower()):
273 continue
274 remote_file_url = urllib.parse.urljoin(url.rstrip("/") + "/", file_path)
275 with requests.get(
276 remote_file_url,
277 auth=self.auth,
278 stream=True,
279 timeout=REQUEST_TIMEOUT_SECONDS,
280 ) as response:
281 if response.status_code != 200:
282 message = f"Failed to download {remote_file_url}: HTTP {response.status_code}"
283 raise OSError(message)
284 local_file_path.parent.mkdir(parents=True, exist_ok=True)
285 _write_atomically(_download_target(base, relative), response)
286 written += 1
287 return written
289 def download_all_files_iterative(
290 self,
291 remote_base_path: str,
292 local_base_path: str,
293 *,
294 extension: str | None = None,
295 ) -> int:
296 """Download a tree with an optional case-insensitive suffix; fail on any error and return the count."""
297 stack: list[str] = [remote_base_path]
298 written = 0
299 global_remote_base_path: str = remote_base_path
301 while stack:
302 current_remote_path: str = stack.pop()
303 logger.debug("Current remote path is: {}", current_remote_path)
305 written += self.download_files(
306 global_remote_base_path,
307 current_remote_path,
308 local_base_path,
309 extension=extension,
310 )
312 folders: list[str] = self.list_folders(current_remote_path)
313 if len(folders) == 0:
314 logger.info(
315 "Found no subfolders for current folder: {}", current_remote_path
316 )
317 for folder in folders:
318 # Folder listings return one encoded component, never a remote path to traverse.
319 relative = _remote_relative_path(
320 folder,
321 _join_remote_url(self.hostname, current_remote_path),
322 )
323 if len(relative) != 1:
324 message = "WebDAV folder is not a direct child of its parent"
325 raise ValueError(message)
326 logger.info(
327 "Found subfolder {} for current folder: {}.",
328 folder,
329 current_remote_path,
330 )
331 relative_folder_path: str = str(
332 PurePosixPath(current_remote_path) / folder
333 )
334 stack.append(relative_folder_path)
335 return written